piscina-contract

Contract-first, type-safe task clients for Piscina.

Piscina is a fast Node.js worker-thread pool. piscina-contract does not replace it, run a second queue, or hide its operational API. It adds one small layer that keeps the task names, inputs, outputs, and worker implementations in sync at compile time.

Why

Piscina's run() API is intentionally generic: the worker module decides what a task means. In a growing TypeScript application, that can leave the caller and worker with duplicated, drifting types. A contract makes that boundary explicit:

• callers only see declared task names and typed inputs/results;
• worker handlers must implement every task with the correct signature;
• Piscina remains available for events, metrics, backpressure, custom queues, named exports, and advanced use cases;
• optional Standard Schema validators can validate data at the runtime boundary.

This package requires Node.js 20 or later and Piscina 5.3 or later.

Installation

pnpm add piscina-contract piscina

piscina is a peer dependency, so your application owns the pool version and its configuration.

Quick start

Define a shared contract:

// contract.ts
import { defineContract, task } from "piscina-contract";

export interface PriceInput {
  hotelId: string;
  nights: number;
}

export interface PriceResult {
  total: number;
  currency: string;
}

export interface ReportInput {
  title: string;
}

export interface ReportResult {
  path: string;
}

export const hotelContract = defineContract({
  calculatePrice: task<PriceInput, PriceResult>({ timeoutMs: 10_000 }),
  generateReport: task<ReportInput, ReportResult>(),
});

Implement every task in the worker module:

// worker.ts
import { implementContract } from "piscina-contract";

import { hotelContract } from "./contract.js";

export default implementContract(hotelContract, {
  calculatePrice(input) {
    return {
      total: input.nights * 1_000,
      currency: "TRY",
    };
  },

  async generateReport(input) {
    return { path: await writeReport(input.title) };
  },
});

Create the client in the main thread:

// main.ts
import { createContractPool } from "piscina-contract";

import { hotelContract } from "./contract.js";

const workers = createContractPool(hotelContract, {
  filename: new URL("./worker.js", import.meta.url),
});

try {
  const result = await workers.tasks.calculatePrice({
    hotelId: "hotel-123",
    nights: 3,
  });

  console.log(result); // { total: 3000, currency: "TRY" }
} finally {
  await workers.close();
}

The compiler rejects unknown tasks, wrong inputs, missing handlers, extra handlers, and incorrect handler results. Both synchronous and asynchronous handlers are supported.

A runnable version is in examples/basic.

Timeouts

A task-level timeout is used when the caller does not provide an override:

const contract = defineContract({
  expensiveTask: task<Input, Output>({ timeoutMs: 5_000 }),
});

await workers.tasks.expensiveTask(input); // 5 seconds
await workers.tasks.expensiveTask(input, { timeoutMs: 500 });
await workers.tasks.expensiveTask(input, { timeoutMs: null }); // disable the default

Timeouts cover input validation, worker execution, and output validation. An expired call rejects with TaskTimeoutError and aborts the Piscina task.

Cancellation

Pass an AbortSignal just as you would to Piscina:

const controller = new AbortController();

const result = workers.tasks.expensiveTask(input, {
  signal: controller.signal,
});

controller.abort("request disconnected");
await result; // rejects with TaskCancelledError

Piscina terminates a worker when an already-running worker task is cancelled. The pool may create a replacement worker according to its normal behavior.

Transfer lists

Transfer lists are forwarded to piscina.run() without copying the underlying buffer:

const buffer = new ArrayBuffer(1024);

const size = await workers.tasks.inspectBuffer(buffer, {
  transferList: [buffer],
});

console.log(size);
console.log(buffer.byteLength); // 0: ownership moved to the worker

Worker results marked with Piscina.move() are also preserved because the dispatcher returns the handler result directly to Piscina.

Optional runtime validation

TypeScript types disappear at runtime. Without schemas, piscina-contract performs only compile-time checking and does not validate JavaScript values.

Tasks optionally accept any Standard Schema V1 compatible schema. No schema package is a production dependency of piscina-contract:

import { z } from "zod";
import { defineContract, task } from "piscina-contract";

const priceInput = z.object({
  hotelId: z.string(),
  nights: z.number().int().positive(),
});

const priceResult = z.object({
  total: z.number(),
  currency: z.string(),
});

const contract = defineContract({
  calculatePrice: task<z.input<typeof priceInput>, z.output<typeof priceResult>>({
    inputSchema: priceInput,
    outputSchema: priceResult,
  }),
});

Input is validated and replaced by the schema's parsed value before the worker receives it. Output is validated after Piscina returns it. Schemas themselves are never cloned or sent to a worker. Validation is opt-in because it has a cost; see Benchmarks.

Inline testing

createInlineClient() exposes the same typed task names without starting worker threads:

const client = createInlineClient(hotelContract, {
  calculatePrice: (input) => ({
    total: input.nights * 1_000,
    currency: "TRY",
  }),
  generateReport: async () => ({ path: "/tmp/report.pdf" }),
});

const result = await client.tasks.calculatePrice({
  hotelId: "hotel-123",
  nights: 3,
});

Inline cancellation can reject an asynchronous operation, but it cannot stop a synchronous CPU-bound handler after that handler has started. JavaScript cannot process the abort event while the main thread is blocked.

Piscina compatibility and escape hatch

The real pool is always available as workers.piscina.

workers.piscina.on("needsDrain", pauseProducer);
workers.piscina.on("drain", resumeProducer);

console.log(workers.piscina.queueSize);
console.log(workers.piscina.completed);
console.log(workers.piscina.histogram);
console.log(workers.piscina.utilization);

The wrapper forwards close() and destroy() directly. The underlying instance keeps Piscina's events, statistics, backpressure state, threads, options, custom task queues, load balancers, resource limits, and disposal APIs.

Named exports and dispatcher envelopes

Contract calls use a default-export dispatcher with an internal { task, input } envelope. A runtime helper cannot dynamically create static ESM named exports, so contract methods intentionally do not expose Piscina's per-call name or filename overrides. This prevents a contract call from accidentally bypassing its dispatcher.

Named exports remain fully usable through the underlying pool:

const result = await workers.piscina.run(rawInput, { name: "specialTask" });

The same worker module may export both the default contract dispatcher and ordinary named handlers.

Custom task queues

An envelope changes the raw value visible to a custom task queue. Use queueOptions to attach scheduling metadata through Piscina's official queueOptionsSymbol protocol:

await workers.tasks.calculatePrice(input, {
  queueOptions: { priority: 10 },
});

The metadata is available to the configured TaskQueue; it is not cloned into the worker request. Queues that inspect arbitrary properties on the original raw input instead of queueOptionsSymbol must be adapted, or those calls can use workers.piscina.run() directly.

Piscina comparison

| Capability | Piscina | piscina-contract | | ------------------------------------ | ---------------- | ---------------------------------------- | | Worker pool and scheduling | Owns it | Uses Piscina unchanged | | Type-safe task names | Manual | Inferred from the contract | | Typed task input/output | Generic per pool | Per task | | Complete worker implementation check | Manual | Compile-time and runtime checks | | Runtime validation | User-defined | Optional Standard Schema boundary | | Events, metrics, backpressure | Native | Available through .piscina | | Custom task queues | Native | Preserved; metadata forwarded explicitly | | Named exports | Native | Available through .piscina.run() | | Cancellation and transferables | Native | Forwarded by typed task methods | | Persistent jobs/retries | No | No |

Performance

The schema-free fast path creates one small envelope and returns Piscina's promise directly. It does not create an AbortController, run validation, or add a second queue.

On an Apple Silicon macOS run with Node.js 24.14.0 and Piscina 5.3.0, 20,000 sequential minimal tasks showed approximately 0.44 microseconds (2.69%) per call over direct piscina.run(). Queued tasks retained approximately 39 bytes per task more main-heap memory. Two Standard Schema checks added approximately 1.12 microseconds (6.63%) per call over the schema-free client.

This deliberately tiny task is a worst case for relative wrapper overhead. Worker threads are useful for CPU-heavy tasks, where a sub-microsecond client cost is normally diluted by task runtime. Results vary by machine and workload; run pnpm benchmark in the target environment and read the full methodology and report.

When to use worker threads

Use Piscina for synchronous, CPU-intensive work that would otherwise block the Node.js event loop. Moving ordinary asynchronous I/O into worker threads usually adds overhead without improving throughput.

piscina-contract is not BullMQ, Temporal, Redis, a distributed worker system, a persistent queue, a workflow engine, or a retry system. Process exit loses queued work, exactly as it does with Piscina.

CommonJS

The package publishes ESM, CommonJS, and declaration files:

const { defineContract, task } = require("piscina-contract");

Workers and application output still need to follow Piscina's normal ESM/CommonJS loading rules.

Roadmap

• validate against new Piscina releases and supported Node.js versions;
• collect benchmark results from more operating systems and CPU architectures;
• explore additional schema ergonomics without taking a runtime dependency;
• consider diagnostics hooks only where they preserve the direct Piscina fast path.

Distributed execution, persistence, retries, decorators, and worker hot reload are intentionally out of scope.

Contributing

Bug reports, compatibility fixtures, performance measurements, and focused pull requests are welcome. Read CONTRIBUTING.md before submitting a change.

License

MIT