n-task

A TypeScript library for task parallelization in both backend (Node.js worker threads) and frontend (browser web workers) environments. You define a worker with ordinary methods, spin up a pool of them, and call those methods from your main thread with a promise-based API.

Features

• Cross-platform: Node.js worker threads (@nivinjoseph/n-task/backend) and browser web workers (@nivinjoseph/n-task/frontend) with the same pool API.
• Pooling & queueing: Create N workers; calls are dispatched to the first idle worker, and queued when all are busy.
• Promise-based: invoke() returns a promise that resolves with the method's return value (sync or async methods both work).
• Robust failures: a thrown method, a missing method, a crashed worker, or a disposed pool all reject the corresponding invoke() rather than hanging. Crashed workers are replaced automatically.
• TypeScript first: ships type definitions; invoke<T>() is generic over the return type.

Installation

npm install @nivinjoseph/n-task
# or
yarn add @nivinjoseph/n-task

Requirements

• ESM only. This package is "type": "module" and exposes no CommonJS build — consume it from an ES module (import), not require.
• Backend: Node.js >= 20.
• Frontend: a browser with Web Worker support, and a bundler (Vite, webpack 5, etc.) to build the worker entry.

The package root is intentionally not importable. Always import from the ./backend or ./frontend subpath:

import { TaskPool, TaskWorker } from "@nivinjoseph/n-task/backend";  // Node
import { TaskPool, TaskWorker } from "@nivinjoseph/n-task/frontend"; // browser

How it works

• A TaskWorker is a class you write that runs inside the worker. Each public method becomes a remotely-callable task. You instantiate it once at the bottom of the worker module so it starts listening.
• A TaskPool runs in your main thread, spawns count workers, and exposes invoke(methodName, ...args) to call a worker method on the next available worker.
• Arguments and return values cross the worker boundary via structured clone, so they must be cloneable (primitives, plain objects/arrays, ArrayBuffer, Map, Set, etc.) — not functions or class instances with behavior.

Backend (Node.js)

1. The worker module — math-worker.ts (compiles to math-worker.js):

import { TaskWorker } from "@nivinjoseph/n-task/backend";

class MathWorker extends TaskWorker
{
    // A synchronous task.
    public fib(n: number): number
    {
        return n < 2 ? n : this.fib(n - 1) + this.fib(n - 2);
    }

    // An async task — the pool awaits the returned promise.
    public async hash(value: string): Promise<string>
    {
        const { createHash } = await import("node:crypto");
        return createHash("sha256").update(value).digest("hex");
    }
}

// Instantiate once so the worker starts handling messages.
new MathWorker();

2. The main thread:

import { TaskPool } from "@nivinjoseph/n-task/backend";
import { fileURLToPath } from "node:url";

// The backend pool takes a path to the COMPILED worker file (.js).
const workerPath = fileURLToPath(new URL("./math-worker.js", import.meta.url));

const pool = new TaskPool(workerPath, 4); // 4 worker threads

await pool.initializeWorkers();

// Calls fan out across the 4 workers and run in parallel.
const results = await Promise.all([
    pool.invoke<number>("fib", 40),
    pool.invoke<number>("fib", 41),
    pool.invoke<number>("fib", 42)
]);

console.log(results); // [102334155, 165580141, 267914296]

await pool.dispose();

Frontend (browser)

1. The worker module — image-worker.ts:

import { TaskWorker } from "@nivinjoseph/n-task/frontend";

class ImageWorker extends TaskWorker
{
    public async resize(bytes: ArrayBuffer, width: number): Promise<ArrayBuffer>
    {
        // ...resize logic...
        return resized;
    }
}

// `self` is the web worker global scope.
new ImageWorker(self as unknown as Worker);

2. The main thread:

The frontend pool needs a zero-argument constructor that produces a Worker. With a bundler that understands new URL(..., import.meta.url) (Vite, webpack 5), wrap your worker entry:

import { TaskPool } from "@nivinjoseph/n-task/frontend";

class ImageWorkerClient extends Worker
{
    public constructor()
    {
        super(new URL("./image-worker.js", import.meta.url), { type: "module" });
    }
}

const pool = new TaskPool(ImageWorkerClient, navigator.hardwareConcurrency ?? 4);

await pool.initializeWorkers();

const resized = await pool.invoke<ArrayBuffer>("resize", bytes, 256);

await pool.dispose();

n-task only requires that new TaskWorkerClass() yield a Worker instance. How you obtain that constructor (the new URL wrapper above, a ?worker import, a worker-loader, etc.) depends on your bundler.

One-time worker initialization

If each worker needs setup before handling tasks (open a DB handle, load a model, etc.), give it an initializer method and pass its name to initializeWorkers. It runs once on every worker before the pool accepts invoke calls. The initializer and its params are retained for the pool's lifetime so replacement workers can be initialized the same way — pass handles and config, not large one-shot buffers:

class DbWorker extends TaskWorker
{
    private _client: Client | null = null;

    public async connect(connectionString: string): Promise<void>
    {
        this._client = await Client.connect(connectionString);
    }

    public async query(sql: string): Promise<Array<unknown>>
    {
        return this._client!.query(sql);
    }
}

await pool.initializeWorkers("connect", process.env.DB_URL);
const rows = await pool.invoke<Array<unknown>>("query", "select 1");

Error handling

invoke<T>() rejects (rather than hanging) in all of these cases:

| Situation | Rejection | |---|---| | The worker method throws (sync or async) | An Error carrying the original message and stack | | The method name isn't implemented on the worker | Error: Method '<name>' not implemented in TaskWorker '<class>' | | The worker thread crashes or exits unexpectedly | An error describing the crash; the pool spawns a replacement, which must answer a readiness probe (and run the initializer, if given) before it receives work | | The pool cannot keep any worker alive (see below) | ApplicationException; the pool is faulted and disposed, and pool.isFaulted is true | | dispose() is called while the task is queued or in flight | ObjectDisposedException |

try
{
    await pool.invoke("query", "bad sql");
}
catch (e)
{
    console.error("task failed:", e); // a real Error with message + stack
}

Worker recovery

The pool tries to keep count workers alive. Every new worker, at startup or as a replacement, must first answer a readiness probe and then, if you passed an initializer to initializeWorkers, run it — all within the bring-up deadline (bringUpTimeoutMs, default 60 s), so a worker that loads but never answers cannot hold its slot forever. Only workers that have completed this handshake receive tasks. When a worker crashes or exits, only the task it was running is rejected; the worker is removed and a replacement is spawned. Consecutive losses and accumulated boot failures back the respawns off exponentially (up to 500 ms between attempts; the loss count resets as soon as any task completes, boot failures only once a new worker proves itself), so a transient cause cannot burn through the recovery budget in an instant. Queued tasks wait for a ready worker and are never handed to one that has not proven itself.

Startup is all-or-nothing: if any worker fails to load, fails its initializer, or dies before every worker is ready, initializeWorkers rejects with that error and the pool is disposed.

After startup, a lost worker — original or replacement — counts as a boot failure unless it had completed at least one task or was running one when it died. That covers a module that fails to load, a constructor that throws, an initializer that throws, and a worker that dies idle right after starting. A crash while running a real task is never a boot failure, however early it happens. After TaskPool.maxBootFailures (10) consecutive boot failures, with no worker spawned after the most recent boot failure completing a task in between, the pool stops replacing lost replacements. A worker that had completed a task, or died in the middle of one, still gets one fresh replacement attempt when it is lost, so capacity can recover once the cause is fixed. One trade-off follows from this: a worker module that boots, takes a queued task, and dies shortly after looks exactly like a healthy worker crashing on a poison task, so while tasks keep arriving the pool favors availability and keeps replacing it — each affected task rejects promptly rather than the pool faulting.

• If the cap trips while other workers are alive, the pool keeps serving on them at reduced capacity. pool.liveWorkerCount tells you how many workers are alive, so you can decide whether to rebuild.
• If no live worker remains and none will be spawned, the pool faults: queued tasks are rejected with an ApplicationException, every later invoke() rejects immediately, and pool.isFaulted is true. Create a new pool to recover.

In browsers, an uncaught exception does not stop a worker, so the worker answers it itself: an uncaught error or unhandled rejection while a task (or the initializer) is in flight may be the very thing that would have completed it, so that call is rejected with the real error and the worker keeps running; one that surfaces while the worker is idle (a stray timer callback, say) is logged inside the worker and otherwise ignored. Only an error while the worker module is still loading reaches the pool, which treats it as a failed bring-up. For workers built against older versions of this library the pool falls back to deciding from the worker's state: ignored when idle and proven, fatal (reject the in-flight call and replace the worker) while loading, initializing or running a task.

API reference

TaskPool

| | Backend | Frontend | |---|---|---| | Constructor | new TaskPool(taskWorkerFile: string, count = 1, bringUpTimeoutMs?: number) | new TaskPool(taskWorker: Function, count = 1, bringUpTimeoutMs?: number) | | taskWorkerFile / taskWorker | Path to the compiled worker .js file | A zero-arg constructor that returns a Worker | | count | Number of workers to spawn (default 1) | Number of workers to spawn (default 1) | | bringUpTimeoutMs | Deadline for a worker's readiness probe plus initializer (default 60_000, max 2_147_483_647) | Same |

Methods

initializeWorkers(initializerMethod?: string, ...initializerParams: Array<any>): Promise<void>

Spawns the workers and waits until each has loaded and answered a readiness probe. If initializerMethod is provided, also calls it once on every worker before resolving. If any worker fails to load, fails to initialize, dies, or misses the bring-up deadline (bringUpTimeoutMs) before all are ready, the pool is disposed and the error is rethrown. Throws if already started or disposed.

invoke<T>(method: string, ...params: Array<any>): Promise<T>

Runs method(...params) on the next available worker (queueing if all are busy) and resolves with its return value. Throws if the pool isn't initialized or is disposed.

dispose(): Promise<void>

Rejects all queued and in-flight tasks, terminates every worker, and resolves once teardown is complete. Idempotent.

Properties

readonly isFaulted: boolean

true once the pool has given up: no live workers remain and none will be spawned (see Worker recovery). Every invoke() rejects from then on; create a new pool to recover.

readonly liveWorkerCount: number

Workers currently alive, including any replacement still booting. Equals count in a healthy pool; lower means the pool is running degraded, or a replacement spawn is briefly pending its backoff delay (see Worker recovery). 0 once disposed.

static readonly maxBootFailures: number

Consecutive boot failures after which the pool stops replacing lost replacements. Currently 10; see Worker recovery.

TaskWorker

Abstract base class you extend to define a worker. Each public method is callable via pool.invoke("methodName", ...). Methods may be synchronous or async; the return value is sent back to the caller.

• Backend: new MyWorker() at the bottom of the worker module (no constructor argument).
• Frontend: new MyWorker(self as unknown as Worker) at the bottom of the worker module.

Development

Requires Node.js >= 20. A pinned Yarn release ships in the repo, so corepack enable is all you need.

yarn install        # install dependencies
yarn ts-build       # type-check + lint the library
yarn test           # backend tests (Node worker threads)
yarn test-browser   # frontend smoke test in a real browser

yarn test-browser is self-contained: it builds the library, downloads a headless Chromium on first run (via Playwright, cached afterwards), starts a Vite dev server to bundle the worker, and drives the frontend TaskPool end-to-end in the browser.

On Linux CI you may also need the browser's system libraries:

yarn playwright install --with-deps chromium

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

Support

For support, please open an issue in the GitHub repository.

License

MIT License - See LICENSE file for details.