Workers

Parallel computation and isolated event loops. Declared entries are bundled into the application image; workers do not compile or load source files at runtime. Native executors and GC helpers are shared across JavaScript isolates.

Import from maligator:workers. This API is experimental. Load declarations as described in Use TypeScript. Platform imports do not require a surface flag. Start with Run tasks in workers for a complete pool example. Use messages and transfers for buffer ownership, or cancellation and shutdown for lifecycle handling.

Task submissions can throw before admission: invalid task names or argument tuples (TypeError), a closed pool (InvalidStateError), saturation (QueueFullError), clone/transfer validation (DataCloneError), and an already-aborted signal's reason. Once admitted, execution and cancellation settle through the returned promise. A failed worker does not replay accepted work.

A running task must cooperate with cancellation. close() drains work; terminate() rejects outstanding work and waits for running tasks to settle and workers to be joined. Pool queue defaults differ from direct worker/channel message defaults; use the corresponding option type below.

export const createPool: <Module>(entry: WorkerUrl<Module>, options?: PoolOptions) => WorkerPool<Module>;

import { createPool, createWorkerUrl } from "maligator:workers";

const tasks = createWorkerUrl<typeof import("./tasks.ts")>(
	"./tasks.ts",
	import.meta.url,
);
const pool = createPool(tasks, { size: 2, maxQueuedTasks: 4 });

try {
	await pool.ready;
	console.log(await pool.run("sum", [[1, 2, 3]]));

	const inputs: Array<[Array<number>]> = [[[1, 2]], [[3, 4]]];
	for await (const total of pool.map("sum", inputs, { window: 2 })) {
		console.log(total);
	}

	const controller = new AbortController();
	const pending = pool.run("waitForCancellation", [], {
		signal: controller.signal,
	});
	const cancelled = pending.then(
		() => false,
		(reason: unknown) => reason === controller.signal.reason,
	);
	controller.abort();
	console.log(await cancelled);
} finally {
	await pool.close();
}

import type { TaskContext } from "maligator:workers";

export function sum(context: TaskContext, values: Array<number>): number {
	let total = 0;
	for (const value of values) {
		context.throwIfCancelled();
		total += value;
	}
	return total;
}

export async function waitForCancellation(context: TaskContext): Promise<void> {
	context.throwIfCancelled();
	await new Promise<void>((resolve) => {
		context.signal.addEventListener("abort", () => resolve(), { once: true });
	});
	context.throwIfCancelled();
}

export const createWorkerUrl: <Module = unknown>(specifier: string, base: string) => WorkerUrl<Module>;

import { createWorkerUrl } from "maligator:workers";

export const tasks = createWorkerUrl<typeof import("./sum.ts")>(
	"./sum.ts",
	import.meta.url,
);

console.log(Object.isFrozen(tasks), tasks.href.startsWith("file:"));

import type { TaskContext } from "maligator:workers";

export function sum(context: TaskContext, values: Array<number>): number {
	context.throwIfCancelled();
	return values.reduce((total, value) => total + value, 0);
}

export const transfer: <Value>(value: Value, transfer: ReadonlyArray<Transferable>) => TransferResult<Value>;

import { transfer } from "maligator:workers";
import type { TaskContext } from "maligator:workers";

export function reverse(context: TaskContext, buffer: ArrayBuffer) {
	context.throwIfCancelled();
	new Uint8Array(buffer).reverse();
	return transfer(buffer, [buffer]);
}

import { createPool, createWorkerUrl } from "maligator:workers";

const tasks = createWorkerUrl<typeof import("./transfer-task.ts")>(
	"./transfer-task.ts",
	import.meta.url,
);
const pool = createPool(tasks, { size: 1 });

try {
	await pool.ready;
	const bytes = new Uint8Array([1, 2, 3]);
	const pending = pool.run("reverse", [bytes.buffer], {
		transfer: [bytes.buffer],
	});
	console.log(bytes.byteLength);
	const result = new Uint8Array(await pending);
	console.log(Array.from(result).join(","));
} finally {
	await pool.close();
}

export const Worker: { new<Send = unknown, Receive = unknown>(entry: WorkerUrl, options?: WorkerOptions): Worker<Send, Receive> };

import { createWorkerUrl, Worker } from "maligator:workers";

const entry = createWorkerUrl("./echo.ts", import.meta.url);
const worker = new Worker<string, string>(entry);

try {
	const reply = new Promise<string>((resolve) => {
		worker.port.onmessage = (event) => resolve(event.data);
	});
	worker.port.start();
	await worker.ready;
	worker.port.postMessage("workers");
	console.log(await reply);
} finally {
	const exit = await worker.terminate();
	console.log(exit.reason);
}

import { parentPort } from "maligator:workers";

if (parentPort === null) throw new Error("Run this module as a worker");
const port = parentPort;
port.onmessage = (event) => {
	port.postMessage(String(event.data));
};
port.start();

export const MessageChannel: { new(options?: MessageChannelOptions): MessageChannel };

import { MessageChannel, MessagePort } from "maligator:workers";

const channel = new MessageChannel();
try {
	console.log(channel.port1 instanceof MessagePort);
	const received = new Promise<unknown>((resolve) => {
		channel.port2.onmessage = (event) => resolve(event.data);
	});
	channel.port2.start();
	channel.port1.postMessage({ answer: 42 });
	console.log(JSON.stringify(await received));
} finally {
	channel.port1.close();
	channel.port2.close();
}

export const MessagePort: { readonly prototype: MessagePort };

import { MessageChannel, type MessagePort } from "maligator:workers";

function installResponder(port: MessagePort): void {
	port.onmessage = (event) => port.postMessage("reply:" + String(event.data));
	port.start();
}

const { port1, port2 } = new MessageChannel();
try {
	installResponder(port2);
	const reply = new Promise<unknown>((resolve) => {
		port1.onmessage = (event) => resolve(event.data);
	});
	port1.start();
	port1.postMessage("hello");
	console.log(await reply);
} finally {
	port1.close();
	port2.close();
}

export const receiveMessageOnPort: <Receive>(port: MessagePort<unknown, Receive>) => { message: Receive } | undefined;

import { MessageChannel, receiveMessageOnPort } from "maligator:workers";

const channel = new MessageChannel();
try {
	channel.port1.postMessage("first");
	channel.port1.postMessage("second");
	console.log(receiveMessageOnPort(channel.port2)?.message);
	console.log(receiveMessageOnPort(channel.port2)?.message);
	console.log(receiveMessageOnPort(channel.port2));
} finally {
	channel.port1.close();
	channel.port2.close();
}

export const capabilities: () => { readonly threads: boolean; readonly sharedMemory: boolean; readonly parallelism: number; readonly maxWorkers: number };

import { capabilities } from "maligator:workers";

const host = capabilities();
console.log(host.threads, host.sharedMemory);
console.log(host.parallelism >= 1, host.maxWorkers >= 1);

export const parentPort: MessagePort | null;

import { parentPort } from "maligator:workers";

if (parentPort === null) throw new Error("Run this module as a worker");
const port = parentPort;
port.onmessage = (event) => {
	port.postMessage(String(event.data));
};
port.start();

export const workerData: unknown;

import { workerData } from "maligator:workers";

function readConfig(value: unknown): { label: string } {
	if (
		typeof value !== "object" ||
		value === null ||
		!("label" in value) ||
		typeof value.label !== "string"
	) {
		throw new TypeError("Expected worker data with a string label");
	}
	return { label: value.label };
}

const config = readConfig(workerData);
console.log(config.label);

import { createWorkerUrl, Worker } from "maligator:workers";

const entry = createWorkerUrl("./worker-data.ts", import.meta.url);
const worker = new Worker(entry, { data: { label: "thumbnail" } });
try {
	await worker.ready;
	const exit = await worker.closed;
	if (exit.code !== 0) throw new Error("Worker failed");
} finally {
	await worker.terminate();
}

WorkerUrl

An immutable image-local worker entry declaration. The compiler resolves the declaration independently of how libraries pass the descriptor or its href onward. Erased module types express the caller's assertion; runtime entry identity is validated.

type WorkerUrl<Module = unknown> = { readonly href: string; readonly __workerModule?: Module };

WorkerUrl.href

readonly href: string;

WorkerUrl.__workerModule

readonly __workerModule?: Module

Transferable

ArrayBuffer stores move at admission. MessagePort endpoints transfer ownership. SharedArrayBuffer is cloned by sharing its backing and cannot be transferred.

type Transferable = ArrayBuffer | MessagePort;

Types: MessagePort

TaskContext

Cancellation is cooperative while a pool task runs. The worker remains occupied until the task's returned promise settles.

type TaskContext = {
 /** Worker-local cancellation signal. Cancellation does not preempt JavaScript. */
 readonly signal: AbortSignal;
 /** Throws if cancelled. Check between bounded pieces of work. */
 throwIfCancelled(): void;
};

TaskContext.signal

Worker-local cancellation signal. Cancellation does not preempt JavaScript.

readonly signal: AbortSignal;

TaskContext.throwIfCancelled

Throws if cancelled. Check between bounded pieces of work.

throwIfCancelled(): void;

TransferResult

An opaque result envelope. Constructing it does not detach buffers; publication commits transfers.

type TransferResult<Value> = { readonly value: Value; readonly __transferResult: unique symbol };

TransferResult.value

readonly value: Value;

TransferResult.__transferResult

readonly __transferResult: unique symbol

TaskNames

Names of context-first exported task functions.

type TaskNames<Module> = { [Key in keyof Module]-?: Module[Key] extends (context: TaskContext, ...args: infer Args) => unknown ? Key : never }[keyof Module] & string;

Types: TaskContext

TaskArgs

The task's argument tuple, excluding its local cancellation context.

type TaskArgs<Function> = Function extends (context: TaskContext, ...args: infer Args) => unknown ? Args : never;

Types: TaskContext

TaskValue

The settled task result after unwrapping an explicit transfer envelope.

type TaskValue<Function> = Function extends (...args: Array<never>) => infer Result ? Awaited<Result> extends TransferResult<infer Value> ? Value : Awaited<Result> : never;

Types: TransferResult

PoolOptions

Fixed persistent worker count and admission bounds. Every worker has independent module state. A size-one pool dispatches admitted tasks serially.

type PoolOptions = {
 /** Positive safe integer; defaults to host parallelism, falling back to 1. */
 size?: number;
 /** Nonnegative safe integer; defaults to four times size. Zero disables the waiting queue. */
 maxQueuedTasks?: number;
 /** Positive safe integer in bytes; defaults to 67108864 (64 MiB). */
 maxQueuedBytes?: number;
 /** Positive safe integer in bytes; defaults to maxQueuedBytes. */
 maxMessageBytes?: number;
 /** Worker name prefix; each worker receives a one-based suffix. */
 name?: string;
};

PoolOptions.size

Positive safe integer; defaults to host parallelism, falling back to 1.

size?: number;

PoolOptions.maxQueuedTasks

Nonnegative safe integer; defaults to four times size. Zero disables the waiting queue.

maxQueuedTasks?: number;

PoolOptions.maxQueuedBytes

Positive safe integer in bytes; defaults to 67108864 (64 MiB).

maxQueuedBytes?: number;

PoolOptions.maxMessageBytes

Positive safe integer in bytes; defaults to maxQueuedBytes.

maxMessageBytes?: number;

PoolOptions.name

Worker name prefix; each worker receives a one-based suffix.

name?: string;

RunOptions

The transfer list commits synchronously when run returns normally. Validation, saturation, closed pools and already-aborted signals throw before admission.

type RunOptions = {
 /** Already-aborted signals throw before admission. Running cancellation is cooperative. */
 signal?: AbortSignal;
 /** Ownership moves synchronously on successful admission; rejection leaves values unchanged. */
 transfer?: ReadonlyArray<Transferable>;
};

RunOptions.signal

Already-aborted signals throw before admission. Running cancellation is cooperative.

signal?: AbortSignal;

RunOptions.transfer

Ownership moves synchronously on successful admission; rejection leaves values unchanged.

transfer?: ReadonlyArray<Transferable>;

Types: Transferable

MapOptions

The window bounds pulled inputs and buffered results together. Results are yielded in input order; iterator return cancels this map's work and closes its input.

type MapOptions<Args> = {
 /** Cancels only work owned by this map; running tasks must cooperate. */
 signal?: AbortSignal;
 /** Positive safe integer; defaults to pool size. Bounds pulled inputs and buffered results together. */
 window?: number;
 /** Called for each input before admission to choose values whose ownership moves. */
 transfer?: (args: Args, index: number) => ReadonlyArray<Transferable>;
};

MapOptions.signal

Cancels only work owned by this map; running tasks must cooperate.

signal?: AbortSignal;

MapOptions.window

Positive safe integer; defaults to pool size. Bounds pulled inputs and buffered results together.

window?: number;

MapOptions.transfer

Called for each input before admission to choose values whose ownership moves.

transfer?: (args: Args, index: number) => ReadonlyArray<Transferable>;

Types: Transferable

PoolStats

A snapshot of this pool's scheduling and settled operations.

type PoolStats = { readonly size: number; readonly active: number; readonly queued: number; readonly completed: number; readonly failed: number; readonly cancelled: number };

PoolStats.size

readonly size: number;

PoolStats.active

readonly active: number;

PoolStats.queued

readonly queued: number;

PoolStats.completed

readonly completed: number;

PoolStats.failed

readonly failed: number;

PoolStats.cancelled

readonly cancelled: number

WorkerPool

A bounded task scheduler over isolated persistent workers. Each worker executes one task through asynchronous settlement. No accepted task is replayed after worker failure.

type WorkerPool<Module> = {
 /** Resolves after every worker evaluates the entry; rejects on startup failure. */
 readonly ready: Promise<void>;
 /** Throws on failed admission; returns a promise for the accepted task's result or failure. */
 run<Key extends TaskNames<Module>>(name: Key, args: TaskArgs<Module[Key]>, options?: RunOptions): Promise<TaskValue<Module[Key]>>;
 /** Yields in input order. Returning from the iterator cancels its work and closes its input. */
 map<Key extends TaskNames<Module>>(name: Key, args: Iterable<TaskArgs<Module[Key]>> | AsyncIterable<TaskArgs<Module[Key]>>, options?: MapOptions<TaskArgs<Module[Key]>>): AsyncIterable<TaskValue<Module[Key]>>;
 /** Stops admission, drains accepted tasks, then joins every worker. Idempotent. */
 close(): Promise<void>;
 /** Rejects outstanding tasks with AbortError; joins workers after running tasks cooperate and settle. */
 terminate(): Promise<void>;
 /** Returns a scheduling snapshot; counts are not a subscription. */
 stats(): PoolStats;
 /** Keeps the process alive while pool workers exist; returns this pool. */
 ref(): WorkerPool<Module>;
 /** Allows the process to exit without waiting for this pool; returns this pool. */
 unref(): WorkerPool<Module>;
 /** Reports whether this pool currently keeps the process alive. */
 hasRef(): boolean;
};

WorkerPool.ready

Resolves after every worker evaluates the entry; rejects on startup failure.

readonly ready: Promise<void>;

WorkerPool.run

Throws on failed admission; returns a promise for the accepted task's result or failure.

run<Key extends TaskNames<Module>>(name: Key, args: TaskArgs<Module[Key]>, options?: RunOptions): Promise<TaskValue<Module[Key]>>;

Types: TaskNames, TaskArgs, TaskValue, RunOptions

WorkerPool.map

Yields in input order. Returning from the iterator cancels its work and closes its input.

map<Key extends TaskNames<Module>>(name: Key, args: Iterable<TaskArgs<Module[Key]>> | AsyncIterable<TaskArgs<Module[Key]>>, options?: MapOptions<TaskArgs<Module[Key]>>): AsyncIterable<TaskValue<Module[Key]>>;

Types: TaskNames, TaskArgs, TaskValue, MapOptions

WorkerPool.close

Stops admission, drains accepted tasks, then joins every worker. Idempotent.

close(): Promise<void>;

WorkerPool.terminate

Rejects outstanding tasks with AbortError; joins workers after running tasks cooperate and settle.

terminate(): Promise<void>;

WorkerPool.stats

Returns a scheduling snapshot; counts are not a subscription.

stats(): PoolStats;

Types: PoolStats

WorkerPool.ref

Keeps the process alive while pool workers exist; returns this pool.

ref(): WorkerPool<Module>;

Types: WorkerPool

WorkerPool.unref

Allows the process to exit without waiting for this pool; returns this pool.

unref(): WorkerPool<Module>;

Types: WorkerPool

WorkerPool.hasRef

Reports whether this pool currently keeps the process alive.

hasRef(): boolean;

WorkerExit

A terminal record published only after the native worker is joined and its slot is released.

type WorkerExit = { readonly id: number; readonly code: number; readonly reason: 'completed' | 'terminated' | 'error'; readonly error?: unknown };

WorkerExit.id

readonly id: number;

WorkerExit.code

readonly code: number;

WorkerExit.reason

readonly reason: 'completed' | 'terminated' | 'error';

WorkerExit.error

readonly error?: unknown

WorkerOptions

Worker data is snapshotted before startup. The native host bounds live workers and message admission process-wide.

type WorkerOptions = {
 name?: string;
 /** Cloned into workerData before startup; later caller mutations are not shared. */
 data?: unknown;
 /** Values moved as part of startup-data admission. */
 transfer?: ReadonlyArray<Transferable>;
 /** Positive safe integer, at most 4294967295; defaults to 4096 messages per endpoint. */
 maxQueuedMessages?: number;
 /** Positive safe integer in bytes; defaults to 67108864 (64 MiB) per endpoint. */
 maxQueuedBytes?: number;
 /** Positive safe integer in bytes; defaults to 16777216 (16 MiB) per message. */
 maxMessageBytes?: number;
};

WorkerOptions.name

name?: string;

WorkerOptions.data

Cloned into workerData before startup; later caller mutations are not shared.

data?: unknown;

WorkerOptions.transfer

Values moved as part of startup-data admission.

transfer?: ReadonlyArray<Transferable>;

Types: Transferable

WorkerOptions.maxQueuedMessages

Positive safe integer, at most 4294967295; defaults to 4096 messages per endpoint.

maxQueuedMessages?: number;

WorkerOptions.maxQueuedBytes

Positive safe integer in bytes; defaults to 67108864 (64 MiB) per endpoint.

maxQueuedBytes?: number;

WorkerOptions.maxMessageBytes

Positive safe integer in bytes; defaults to 16777216 (16 MiB) per message.

maxMessageBytes?: number;

MessageChannelOptions

Each endpoint bounds its pending message count and bytes. Rejection leaves the sender's transferables unchanged.

type MessageChannelOptions = {
 /** Positive safe integer, at most 4294967295; defaults to 4096 messages per endpoint. */
 maxQueuedMessages?: number;
 /** Positive safe integer in bytes; defaults to 67108864 (64 MiB) per endpoint. */
 maxQueuedBytes?: number;
 /** Positive safe integer in bytes; defaults to 16777216 (16 MiB) per message. */
 maxMessageBytes?: number;
};

MessageChannelOptions.maxQueuedMessages

Positive safe integer, at most 4294967295; defaults to 4096 messages per endpoint.

maxQueuedMessages?: number;

MessageChannelOptions.maxQueuedBytes

Positive safe integer in bytes; defaults to 67108864 (64 MiB) per endpoint.

maxQueuedBytes?: number;

MessageChannelOptions.maxMessageBytes

Positive safe integer in bytes; defaults to 16777216 (16 MiB) per message.

maxMessageBytes?: number;

MessagePort instance

An ordered bidirectional endpoint with transactional transfer and bounded queues. Message listeners and values are owned by the receiving isolate.

interface MessagePort<Send = unknown, Receive = unknown> extends EventTarget {
 /** Clones and admits a message synchronously. Transfer commits only after successful validation/admission. */
 postMessage(value: Send, transfer?: ReadonlyArray<Transferable>): void;
 /** Setting this handler starts delivery. addEventListener listeners need start(). */
 onmessage: ((event: MessageEvent<Receive>) => void) | null;
 /** Receives message decoding failures in the receiving isolate. */
 onmessageerror: ((event: MessageEvent<unknown>) => void) | null;
 /** Enables delivery to event listeners; repeated calls are harmless. */
 start(): void;
 /** Discards queued messages and closes this endpoint. */
 close(): void;
 /** Keeps the receiving isolate alive; returns this port. */
 ref(): MessagePort<Send, Receive>;
 /** Allows the receiving isolate to exit; returns this port. */
 unref(): MessagePort<Send, Receive>;
 hasRef(): boolean;
}

MessagePort.postMessage

Clones and admits a message synchronously. Transfer commits only after successful validation/admission.

postMessage(value: Send, transfer?: ReadonlyArray<Transferable>): void;

Types: Transferable

MessagePort.onmessage

Setting this handler starts delivery. addEventListener listeners need start().

onmessage: ((event: MessageEvent<Receive>) => void) | null;

MessagePort.onmessageerror

Receives message decoding failures in the receiving isolate.

onmessageerror: ((event: MessageEvent<unknown>) => void) | null;

MessagePort.start

Enables delivery to event listeners; repeated calls are harmless.

start(): void;

MessagePort.close

Discards queued messages and closes this endpoint.

close(): void;

MessagePort.ref

Keeps the receiving isolate alive; returns this port.

ref(): MessagePort<Send, Receive>;

Types: MessagePort

MessagePort.unref

Allows the receiving isolate to exit; returns this port.

unref(): MessagePort<Send, Receive>;

Types: MessagePort

MessagePort.hasRef

hasRef(): boolean;

MessageChannel instance

A standalone channel whose endpoints may be transferred to workers.

type MessageChannel = { readonly port1: MessagePort; readonly port2: MessagePort };

MessageChannel.port1

readonly port1: MessagePort;

Types: MessagePort

MessageChannel.port2

readonly port2: MessagePort

Types: MessagePort

Worker instance

A long-lived isolated module and its parent communication port. Startup completes after module evaluation; shutdown completes after native thread reaping.

type Worker<Send = unknown, Receive = unknown> = EventTarget & {
 readonly id: number;
 /** Resolves after entry evaluation; rejects on startup failure. */
 readonly ready: Promise<void>;
 /** Resolves with a terminal record after the native thread is joined and its slot released. */
 readonly closed: Promise<WorkerExit>;
 readonly port: MessagePort<Send, Receive>;
 /** Requests shutdown and resolves after thread reaping. */
 terminate(): Promise<WorkerExit>;
 ref(): Worker<Send, Receive>;
 unref(): Worker<Send, Receive>;
 hasRef(): boolean;
};

Worker.id

readonly id: number;

Worker.ready

Resolves after entry evaluation; rejects on startup failure.

readonly ready: Promise<void>;

Worker.closed

Resolves with a terminal record after the native thread is joined and its slot released.

readonly closed: Promise<WorkerExit>;

Types: WorkerExit

Worker.port

readonly port: MessagePort<Send, Receive>;

Types: MessagePort

Worker.terminate

Requests shutdown and resolves after thread reaping.

terminate(): Promise<WorkerExit>;

Types: WorkerExit

Worker.ref

ref(): Worker<Send, Receive>;

Types: Worker

Worker.unref

unref(): Worker<Send, Receive>;

Types: Worker

Worker.hasRef

hasRef(): boolean;