Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion lib/services/bulkhead-registry.service.ts
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ import type { BulkheadOptions } from '../interfaces/bulkhead-options.interface.j
import type { ResilienceModuleOptions } from '../interfaces/resilience-module-options.interface.js';
import { BulkheadPolicy } from '../policies/bulkhead.policy.js';
import { RESILIENCE_MODULE_OPTIONS } from '../resilience.module-definition.js';
import { definedFields } from '../utils/options.util.js';
import { ms, NamedRegistry } from './named-registry.service.js';

@Injectable()
Expand All @@ -18,7 +19,7 @@ export class BulkheadRegistry extends NamedRegistry<BulkheadPolicy, BulkheadOpti
}

protected create(name: string, options: BulkheadOptions = {}) {
return new BulkheadPolicy({ ...this.moduleOptions.defaults?.bulkhead, ...options, name });
return new BulkheadPolicy({ ...this.moduleOptions.defaults?.bulkhead, ...definedFields(options), name });
}

protected normalize({ queueTimeout, ...rest }: BulkheadOptions) {
Expand Down
4 changes: 3 additions & 1 deletion lib/services/circuit-breaker-registry.service.ts
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ import type { CircuitBreakerOptions } from '../interfaces/circuit-breaker-option
import type { ResilienceModuleOptions } from '../interfaces/resilience-module-options.interface.js';
import { CircuitBreakerPolicy } from '../policies/circuit-breaker.policy.js';
import { RESILIENCE_MODULE_OPTIONS } from '../resilience.module-definition.js';
import { definedFields } from '../utils/options.util.js';
import { ms, NamedRegistry } from './named-registry.service.js';

@Injectable()
Expand All @@ -18,7 +19,8 @@ export class CircuitBreakerRegistry extends NamedRegistry<CircuitBreakerPolicy,
}

protected create(name: string, options: CircuitBreakerOptions = {}) {
return new CircuitBreakerPolicy({ ...this.moduleOptions.defaults?.circuitBreaker, ...options, name });
const defaults = this.moduleOptions.defaults?.circuitBreaker;
return new CircuitBreakerPolicy({ ...defaults, ...definedFields(options), name });
}

protected normalize({ openDuration, slidingWindow, ...rest }: CircuitBreakerOptions) {
Expand Down
3 changes: 2 additions & 1 deletion lib/services/entrypoint-planner.service.ts
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,8 @@ import {
RETRY_METADATA,
TIMEOUT_METADATA,
} from '../resilience.constants.js';
import { definedFields, mergeRetry, PolicyFactory, type Stage } from './policy-factory.service.js';
import { definedFields } from '../utils/options.util.js';
import { mergeRetry, PolicyFactory, type Stage } from './policy-factory.service.js';

// Metadata keys of other official packages, read without importing them.
const GRAPHQL_RESOLVER_TYPE = 'graphql:resolver_type';
Expand Down
19 changes: 8 additions & 11 deletions lib/services/policy-factory.service.ts
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@ import { ResiliencePolicy } from '../policies/resilience.policy.js';
import { normalizeRetry, RetryPolicy } from '../policies/retry.policy.js';
import { TimeoutPolicy } from '../policies/timeout.policy.js';
import { RESILIENCE_MODULE_OPTIONS } from '../resilience.module-definition.js';
import { underOption } from '../utils/options.util.js';
import { definedFields, underOption } from '../utils/options.util.js';
import { BulkheadRegistry } from './bulkhead-registry.service.js';
import { CircuitBreakerRegistry } from './circuit-breaker-registry.service.js';
import type { NamedRegistry } from './named-registry.service.js';
Expand Down Expand Up @@ -159,7 +159,7 @@ export class PolicyFactory {
if (spec.circuitBreaker !== undefined) {
stages.circuitBreaker = underOption('circuitBreaker', () =>
this.stateful(spec.circuitBreaker!, this.breakers, source, (options) =>
new CircuitBreakerPolicy({ ...defaults.circuitBreaker, ...options, name }),
new CircuitBreakerPolicy({ ...defaults.circuitBreaker, ...definedFields(options), name }),
),
);
}
Expand All @@ -179,7 +179,7 @@ export class PolicyFactory {
if (spec.bulkhead !== undefined) {
stages.bulkhead = underOption('bulkhead', () =>
this.stateful(spec.bulkhead!, this.bulkheads, source, (options) =>
new BulkheadPolicy({ ...defaults.bulkhead, ...options, name }),
new BulkheadPolicy({ ...defaults.bulkhead, ...definedFields(options), name }),
),
);
}
Expand Down Expand Up @@ -245,9 +245,11 @@ function validateStages(
);
check('timeout', spec.timeout, (timeout) => new TimeoutPolicy(timeout));
check('circuitBreaker', spec.circuitBreaker, unlessNamed((options) =>
new CircuitBreakerPolicy({ ...defaults.circuitBreaker, ...options }),
new CircuitBreakerPolicy({ ...defaults.circuitBreaker, ...definedFields(options) }),
));
check('bulkhead', spec.bulkhead, unlessNamed((options) =>
new BulkheadPolicy({ ...defaults.bulkhead, ...definedFields(options) }),
));
check('bulkhead', spec.bulkhead, unlessNamed((options) => new BulkheadPolicy({ ...defaults.bulkhead, ...options })));

if ('outboundRateLimit' in spec) {
check('outboundRateLimit', spec.outboundRateLimit, unlessNamed((options) => new OutboundRateLimitPolicy(options)));
Expand All @@ -269,7 +271,7 @@ function normalizeDefaultRetry(retry: number | RetryOptions | undefined): RetryO
export function mergeRetry<R extends RetryOptions>(base: RetryOptions, override: R): R {
const backoff =
typeof base.backoff === 'object' && typeof override.backoff === 'object'
? { ...base.backoff, ...override.backoff }
? { ...base.backoff, ...definedFields(override.backoff) }
: (override.backoff ?? base.backoff);

const merged = { ...base, ...definedFields(override) } as R;
Expand All @@ -278,8 +280,3 @@ export function mergeRetry<R extends RetryOptions>(base: RetryOptions, override:
}
return merged;
}

/** Drops `undefined` fields, so that `{ attempts: undefined }` doesn't erase an inherited value. */
export function definedFields<T extends object>(value: T): Partial<T> {
return Object.fromEntries(Object.entries(value).filter(([, v]) => v !== undefined)) as Partial<T>;
}
5 changes: 5 additions & 0 deletions lib/utils/options.util.ts
Original file line number Diff line number Diff line change
Expand Up @@ -67,3 +67,8 @@ export function callPredicate<A extends unknown[]>(
}
return !!result;
}

/** @internal Drops `undefined` fields, so that `{ attempts: undefined }` doesn't erase an inherited value. */
export function definedFields<T extends object>(value: T): Partial<T> {
return Object.fromEntries(Object.entries(value).filter(([, v]) => v !== undefined)) as Partial<T>;
}
36 changes: 36 additions & 0 deletions tests/module.spec.ts
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@ import {
Signal,
Timeout,
TimeoutPolicy,
type Duration,
type ResilienceModuleOptions,
type ResilienceOptionsFactory,
type ResiliencePolicy,
Expand Down Expand Up @@ -573,6 +574,41 @@ describe('ResilienceModule', () => {
expect(stages[1]).toMatchObject({ timeout: 1_000 });
});

it('keeps module defaults for fields a preset or create() leaves undefined', async () => {
// As `ConfigService.get()` answers for a setting that is not set.
const config = { get: <T>(_key: string): T | undefined => undefined };
const moduleRef = await boot(
{
defaults: {
circuitBreaker: { minimumCalls: 1, openDuration: '1m' },
bulkhead: { maxConcurrent: 2 },
retry: { attempts: 4, backoff: { delay: '1s', jitter: 'none' } },
},
presets: {
carrier: {
circuitBreaker: { minimumCalls: config.get<number>('CARRIER_MIN_CALLS') },
bulkhead: { maxConcurrent: config.get<number>('CARRIER_MAX_CONCURRENT') },
retry: { backoff: { delay: config.get<Duration>('CARRIER_DELAY') } },
},
},
},
{},
);
const resilience = moduleRef.get(ResilienceService);
expect(resilience.circuitBreaker('carrier').options).toMatchObject({ minimumCalls: 1, openDuration: 60_000 });
expect(resilience.bulkhead('carrier').maxConcurrent).toBe(2);

const retry = (resilience.preset('carrier') as PolicyWrap).policies[0] as RetryPolicy;
const events = recordEvents();
await expect(
retry.execute(() => Promise.reject(new Error('down')), { signal: AbortSignal.timeout(50) }),
).rejects.toThrow();
expect(events[0]).toMatchObject({ type: 'retry', delayMs: 1_000 });

const created = resilience.create({ bulkhead: { maxConcurrent: config.get<number>('MAX_CONCURRENT') } });
expect(((created as PolicyWrap).policies[0] as BulkheadPolicy).maxConcurrent).toBe(2);
});

it('the options are injectable as RESILIENCE_MODULE_OPTIONS, and tests can override them', async () => {
const options = { presets: { carrier: { circuitBreaker: { openDuration: '30s' as const } } } };
expect((await boot(options, {})).get(RESILIENCE_MODULE_OPTIONS)).toEqual(options);
Expand Down
Loading