Recipes

Fleet-Wide Worker Pool

Bound heavy work (video encoding, PDF rendering, ML inference) to N concurrent executions across every replica:

import { Injectable } from '@nestjs/common';
import { WithSemaphore } from '@nestjs-redisx/semaphore';
import { VideoEncoder } from './types';

@Injectable()
export class EncodingService {
  constructor(private readonly encoder: VideoEncoder) {}

  // At most 2 encodes at a time ACROSS EVERY INSTANCE - the fleet-wide
  // analogue of a bounded worker pool.
  @WithSemaphore({ key: 'video:encode', permits: 2, ttl: 600_000 })
  async encode(videoId: string): Promise<void> {
    await this.encoder.encode(videoId);
  }

  // Per-tenant bound via a {n} key template; saturation skips instead of
  // throwing (good for periodic retried work).
  @WithSemaphore({
    key: 'export:{0}',
    permits: 1,
    waitTimeout: 5_000,
    onUnavailable: 'skip',
  })
  async exportTenant(tenantId: string): Promise<void> {
    // runs at most once concurrently per tenant
  }
}

Scale replicas freely - the ceiling stays N.

Protecting a Fragile Dependency

A legacy API or database that tolerates only a handful of concurrent callers:

import { Injectable, Inject } from '@nestjs/common';
import { SEMAPHORE_SERVICE, ISemaphoreService } from '@nestjs-redisx/semaphore';
import { LegacyApiClient } from './types';

@Injectable()
export class ReportService {
  constructor(
    @Inject(SEMAPHORE_SERVICE)
    private readonly semaphore: ISemaphoreService,
    private readonly legacyApi: LegacyApiClient,
  ) {}

  // withPermit: acquire -> run -> always release.
  async fetchReport(tenantId: string): Promise<unknown> {
    return this.semaphore.withPermit('legacy-api', 5, () => this.legacyApi.fetchReport(tenantId));
  }

  // Manual permit lifecycle for long multi-step work.
  async fetchWithManualPermit(tenantId: string): Promise<unknown> {
    const permit = await this.semaphore.acquire('legacy-api', 5, {
      ttl: 60_000,
      waitTimeout: 10_000, // give up if no seat frees up in 10s
    });
    try {
      return await this.legacyApi.fetchReport(tenantId);
    } finally {
      await permit.release();
    }
  }

  // tryAcquire: skip this round instead of waiting.
  async fetchIfCapacity(tenantId: string): Promise<unknown | undefined> {
    const permit = await this.semaphore.tryAcquire('legacy-api', 5);
    if (!permit) {
      return undefined; // at capacity - caller decides what to do
    }
    try {
      return await this.legacyApi.fetchReport(tenantId);
    } finally {
      await permit.release();
    }
  }
}

Combine with the circuit breaker: the semaphore bounds concurrency, the breaker reacts to failures - they solve different halves of the problem. Use rate limiting when the contract is "requests per second" rather than "requests in flight".

Per-Tenant Bounds

A {n} template gives every tenant its own semaphore:

@WithSemaphore({ key: 'export:{0}', permits: 1, onUnavailable: 'skip' })
async exportTenant(tenantId: string) { /* one export per tenant at a time */ }

Skip-When-Saturated Cron

Periodic work that should not queue behind itself:

@Cron(CronExpression.EVERY_MINUTE)
@WithSemaphore({ key: 'warmup', permits: 2, waitTimeout: 1000, onUnavailable: 'skip' })
async warmCaches() { /* at most 2 warmups in flight; extras skip this tick */ }

Draining for Maintenance

Freeze a semaphore by watching it, then force-free seats stuck behind an incident:

import { Injectable, Inject } from '@nestjs/common';
import { SEMAPHORE_SERVICE, ISemaphoreService } from '@nestjs-redisx/semaphore';

@Injectable()
export class SemaphoreAdminService {
  constructor(
    @Inject(SEMAPHORE_SERVICE)
    private readonly semaphore: ISemaphoreService,
  ) {}

  async capacitySnapshot(key: string, permits: number) {
    return {
      used: await this.semaphore.usedPermits(key),
      available: await this.semaphore.availablePermits(key, permits),
    };
  }

  // Operator recovery: free every seat on a wedged semaphore. In-flight
  // holders keep running; their eventual release() throws
  // SemaphorePermitNotHeldError (the honest taken-away signal).
  async unblock(key: string): Promise<number> {
    return this.semaphore.forceReleaseAll(key);
  }
}

forceReleaseAll never interrupts running work - in-flight holders complete, and the seats open immediately. Their eventual release() throws SemaphorePermitNotHeldError (the honest signal the seat was taken away); withPermit and @WithSemaphore absorb it into an error log.