Semaphore middlewares

Initial configuration

To begin using the semaphore middlewares, you'll need to create and configure a SemaphoreFactoryResolver instance:

./samples/semaphore.ts

import { SemaphoreFactoryResolver } from "eridu-tech/semaphore";
import { MemorySemaphoreAdapter } from "eridu-tech/semaphore/memory-semaphore-adapter";
import { RedisSemaphoreAdapter } from "eridu-tech/semaphore/redis-semaphore-adapter";
import { Redis } from "ioredis";

export const semaphoreFactoryResolver = new SemaphoreFactoryResolver({
    adapters: {
        memory: new MemorySemaphoreAdapter(),
        redis: new RedisSemaphoreAdapter(
            new Redis("YOUR_REDIS_CONNECTION_STRING"),
        ),
    },
    defaultAdapter: "memory",
});

await semaphoreFactoryResolver.init();

withSemaphoreFactory middleware

The Semaphore middleware wraps function calls with a distributed semaphore, limiting the number of concurrent executions across processes. Before executing the wrapped function, a slot is acquired on a key derived from the function's arguments. If the maximum number of concurrent slots (limit) has already been reached, the call waits (or fails immediately for non-blocking semaphores) until a slot becomes available.

Usage

./samples/with-semaphore.ts

import { withSemaphoreFactory } from "eridu-tech/semaphore/middlewares";
import { use } from "eridu-tech/middleware";
import { semaphoreFactoryResolver } from "./semaphore.js";

const withSemaphore = withSemaphoreFactory(semaphoreFactoryResolver);

const processFile = async (filePath: string): Promise<void> => {
    // Process file — limited concurrency
    // ... process the file
};

// Wrap with semaphore using the default adapter (`memory`) — max 3 concurrent file processes
const throttledProcess = use(
    processFile,
    withSemaphore({
        key: ([filePath]) => `file-path:${filePath}`,
        limit: 3,
    }),
);

// Wrap with semaphore using a specific adapter (`redis`)
const throttledProcessOnRedis = use(
    processFile,
    withSemaphore.use("redis")({
        key: ([filePath]) => `file-path:${filePath}`,
        limit: 3,
    }),
);

// These will run up to 3 at a time
await Promise.all([
    throttledProcess("/data/file1.json"),
    throttledProcess("/data/file2.json"),
    throttledProcess("/data/file3.json"),
    throttledProcess("/data/file4.json"), // Waits for a slot
]);

await throttledProcessOnRedis("/data/redis-file.json");

Calling withSemaphore() directly uses the resolver's default adapter (memory). Use withSemaphore.use("redis") to run the wrapped function through a specific registered adapter.

info

Here is a complete list of settings for the withSemaphore function.

Settings

Option	Type	Description
key	Invocable<TParameters, string>	A function that produces the semaphore key from the wrapped function's arguments. All consumers using the same key share the same semaphore limit
limit	number	Maximum number of concurrent slots (consumers) allowed for the semaphore key
slotId	Invocable<TParameters, string>	Optional function that produces a unique slot identifier for the current acquisition attempt. Defaults to a UUID (v4)
ttl	ITimeSpan | null	Time-to-live for each acquired slot. null means slots never expire automatically; if omitted the factory's default TTL is used

withHttpSemaphoreError middleware

The withHttpSemaphoreError middleware converts every semaphore error thrown by the wrapped endpoint or downstream middleware into an HttpError, so the request fails with a meaningful status code instead of a generic 500. Errors that are not semaphore errors are rethrown untouched, and the original error is preserved as the cause.

Usage

./samples/with-http-semaphore-error.ts

import { HttpRouter, defaultHttpRouterAdapter } from "eridu-tech/http-router";
import { LimitReachedSemaphoreError } from "eridu-tech/semaphore/contracts";
import { withHttpSemaphoreError } from "eridu-tech/semaphore/middlewares";

const router = new HttpRouter({
    router: defaultHttpRouterAdapter(),
});

router.use(withHttpSemaphoreError());

router.endpoint({
    url: "/workers/1",
    method: ["POST"],
    handler: async () => {
        // A 429 response when no semaphore slot is available
        throw LimitReachedSemaphoreError.create("workers");
    },
});

Status codes

Errors	Status
LimitReachedSemaphoreError	429
FailedRefreshSemaphoreError, FailedReleaseSemaphoreError	409
UnableToAcquireSemaphoreError, UnableToReleaseSemaphoreError, UnableToForceReleaseAllSemaphoreError, UnableToRefreshSemaphoreError, UnableToGetStateSemaphoreError	503

A 429 response means the semaphore has no free slot left. A 409 response means the caller's slot is no longer valid. A 503 response means the semaphore adapter failed, so the same request can be retried later. The original error is preserved as the cause, and no identifiers are exposed in the response body.

Further information

For further information refer to eridu-tech/semaphore API docs.