Bulkhead — cockatiel bulkhead(limit) в NestJS

Опирается на правила: R-RES-BH-1 … R-RES-BH-3 и R-RES-BH-X1 — раздел 6. Bulkhead.

Важно знать

• bulkhead(maxConcurrent, queueLimit) из cockatiel — обязательный слой отдельно от connection pool. Pool ограничивает TCP-соединения; bulkhead — одновременные вызовы на уровне event loop.
• В Node нет потоков — bulkhead работает в том же async-контексте. AsyncLocalStorage (trace, MDC) не теряется. Это и есть semaphore-семантика без накладных расходов.
• Выносить outbound I/O в worker_threads/piscina «как bulkhead» — запрещено: AsyncLocalStorage не пробрасывается, treads лишние для I/O-bound.
• Sizing: maxConcurrent ≈ connections × 0.8. Bulkhead должен срабатывать раньше исчерпания pool'а.
• queueLimit — маленький или 0. При queueLimit > 0 backpressure моделируется очередью Promise'ов; без верхней границы это бесконечная очередь, и fail-fast теряется.
• При переполнении cockatiel кидает BulkheadRejectedError — адаптер маппит его так же, как BrokenCircuitError от CB.
• Policy создаётся один раз в DI-провайдере (singleton) — не пересоздаётся на каждый вызов.

Если timeout — «один зависший вызов не висит вечно», а Circuit Breaker — «после N ошибок не пускаем новые», то Bulkhead — «не более N одновременных в принципе». Три слоя работают вместе: timeout ограничивает каждый вызов, bulkhead — их одновременное число, CB — останавливает поток при явной деградации.

Bulkhead и connection pool — два разных слоя

R-RES-BH-1: bulkhead — это не дубль undici connections. Это другой уровень защиты.

При залипании вызова (внешняя система отвечает медленно, но CB ещё не открыт) pool забит на долго — секунды headersTimeout/bodyTimeout. Bulkhead отказывает новым вызовам немедленно, как только счётчик активных промисов достигает лимита.

Пример с доменом Order: pool connections: 10, bulkhead maxConcurrent: 8. При 8 одновременных запросах к СберЭквайрингу 9-й получает BulkheadRejectedError мгновенно. Два свободных TCP-слота остаются как буфер — соединения возвращаются в idle, готовы к следующим вызовам.

Настройка и DI-провайдер

R-RES-BH-2/R-RES-CFG-1: policy собирается из конфига в DI-фабрике и живёт как singleton. Не создаётся внутри метода адаптера — иначе у каждого вызова свой счётчик, изоляция теряется.

// sber-resilience.provider.ts
import { bulkhead, circuitBreaker, CountBreaker, ExponentialBackoff,
         retry, timeout, TimeoutStrategy, wrap, handleType } from 'cockatiel';

export const SBER_POLICY = Symbol('SBER_POLICY');

export const SberPolicyProvider = {
  provide: SBER_POLICY,
  inject: [SberClientConfig],
  useFactory: (cfg: SberClientConfig) =>
    wrap(
      retry(handleType(SberTransientError), {
        maxAttempts: 3,
        backoff: new ExponentialBackoff(),
      }),
      circuitBreaker(handleAll, {
        halfOpenAfter: 30_000,
        breaker: new CountBreaker({ threshold: 0.5, size: 50 }),
      }),
      bulkhead(cfg.maxConcurrent, cfg.queueLimit),   // maxConcurrent ≈ connections × 0.8
      timeout(cfg.callTimeoutMs, TimeoutStrategy.Aggressive),
    ),
};

// sber-client.config.ts (class-validator)
export class SberClientConfig {
  @IsInt() @Min(1) connections!: number;             // undici Agent connections
  @IsInt() @Min(1) maxConcurrent!: number;           // bulkhead: ≈ connections × 0.8
  @IsInt() @Min(0) queueLimit!: number;              // 0 = немедленный fail, N = короткая очередь
  @IsInt() @Min(1) callTimeoutMs!: number;
}

# application.yml
client:
  sber:
    connections: 10
    max-concurrent: 8       # 10 × 0.8
    queue-limit: 2          # небольшая очередь, не бесконечная
    call-timeout-ms: 5000

R-RES-CFG-X1: числа прямо в bulkhead(8) без конфига — скрытая конфигурация, не управляется через внешний config-store.

Использование в out-adapter

R-RES-OAS-1: policy оборачивает public-метод out-adapter, не сгенерированный клиент (SberApi), не handler.

// sber.adapter.ts
@Injectable()
export class SberAdapter implements PaymentPort {
  constructor(
    @Inject(SBER_CLIENT) private readonly agent: Agent,
    @Inject(SBER_POLICY) private readonly policy: ReturnType<typeof wrap>,
  ) {}

  async registerOrder(cmd: RegisterOrderCommand): Promise<OrderRegistration> {
    try {
      const raw = await this.policy.execute(() =>
        request(this.agent, {
          path: '/payment/rest/register.do',
          method: 'POST',
          body: toSberRequest(cmd),
        }),
      );
      return toOrderRegistration(await raw.body.json());   // mapper DTO → domain (R-RES-OAS-4)
    } catch (e) {
      if (e instanceof BulkheadRejectedError || e instanceof BrokenCircuitError)
        throw PaymentPortError.systemUnavailable('sber', e);
      if (e instanceof TaskCancelledError)
        throw PaymentPortError.timeout('sber', e);
      throw PaymentPortError.from(e);
    }
  }

  async findOrder(ref: OrderRef): Promise<OrderStatus> {   // read → retry допустим (R-RES-RE-1)
    try {
      const raw = await this.policy.execute(() =>
        request(this.agent, { path: `/payment/rest/getOrderStatus.do?orderId=${ref.id}`, method: 'GET' }),
      );
      return toOrderStatus(await raw.body.json());
    } catch (e) {
      if (e instanceof BulkheadRejectedError || e instanceof BrokenCircuitError)
        throw PaymentPortError.systemUnavailable('sber', e);
      throw PaymentPortError.from(e);
    }
  }
}

BulkheadRejectedError и BrokenCircuitError маппятся одинаково: система временно недоступна → 503 Service Unavailable или fallback в task-queue. Различие в причине (перегрузка vs деградация) — в метриках и логе, не в HTTP-ответе.

AsyncLocalStorage и semaphore-семантика

R-RES-BH-2: в Java semaphore-bulkhead нужен явно, потому что thread-pool bulkhead создаёт отдельные потоки и теряет MDC/SecurityContext. В Node потоков нет — event loop однопоточный. Cockatiel bulkhead() — это счётчик активных промисов, выполняющихся в одном и том же async-контексте.

// AsyncLocalStorage сохраняется — traceId из middleware доступен внутри policy.execute()
this.policy.execute(async () => {
  const traceId = traceStorage.getStore()?.traceId;   // ← доступен, контекст не потерян
  return this.client.request({ path: `/products/${id}`, method: 'GET' });
});

R-RES-BH-X1: выносить outbound HTTP в worker_threads или piscina-пул «для изоляции» — неверно. Worker threads прерывают AsyncLocalStorage-контекст, лишние треды для I/O-bound нагрузки не дают выигрыша — Node I/O асинхронен по природе. Cockatiel-счётчик достаточен.

Sizing — connections × 0.8

R-RES-BH-3: maxConcurrent = connections × 0.8. Запас 20% гарантирует, что bulkhead срабатывает раньше исчерпания пула.

Для Customer-сервиса с тремя внешними системами:

Суммарный пул: 10 + 6 + 4 = 20 TCP-соединений. Суммарный bulkhead: 8 + 5 + 3 = 16 одновременных вызовов. Выполняется R-RES-ISO-2: сумма соединений ≤ половина пула БД.

queueLimit: 0 — жёсткий fail-fast: 9-й вызов к Sber немедленно получает BulkheadRejectedError. queueLimit: N > 0 — короткая очередь: N вызовов ждут освобождения слота. Разумно при коротких вызовах (<200ms), где слот освобождается быстро. Большое queueLimit — это бесконечная очередь промисов: memory leak при деградации.

Что запрещено

Куда дальше

• Per-system isolation — undici Agent, sizing connections, DI-токены.
• Circuit Breaker — CountBreaker, BrokenCircuitError, half-open.
• Fallback — что делать при BulkheadRejectedError.
• Конфигурация — class-validator, per-system конфиг, фабрика policy.
• Observability — prom-client, onFailure/onBreak, gauge executionSlots.
• Timeouts — cockatiel timeout(), иерархия connect/headers/body/total.
• Retry — ExponentialBackoff, идемпотентность, порядок wrap.
• Async и polling — task-queue вместо sleep-цикла.
• Health checks — @nestjs/terminus, TTL-кеш.
• Где какая защита — outbound vs inbound vs schedulers.
• OpenAPI generator — openapi-typescript, policy на adapter, не на клиенте.