@yaebal/workers

a typed worker_threads pool — offload cpu-heavy bits, keep the bot on the main loop

install

terminal

pnpm add @yaebal/workers

when to reach for it

most handlers are i/o-bound and already served by @yaebal/runner's concurrency. threads help only for genuinely cpu-heavy work: image processing, crypto, heavy parsing. offload that — not the whole bot.

usage

register<Tasks>(handlers) inside a worker file declares the named tasks the pool can call. createPool<Tasks>(file, options?) spawns the workers; share the Tasks type between the two and pool.run(name, arg?, options?) fully infers the argument and result types per task. tasks queue centrally and dispatch to the least-busy ready worker — a slow task never blocks an idle one.

tasks.ts

// tasks.ts — runs in a worker thread
import { register } from "@yaebal/workers";

export type Tasks = {
  resize: (buf: Uint8Array) => Promise<Uint8Array>;
  hash: (s: string) => string;
};

register<Tasks>({
  resize: (buf) => sharp(buf).resize(100).toBuffer(),
  hash:   (s)   => crypto.createHash("sha256").update(s).digest("hex"),
});

bot.ts

// bot — stays on the main event loop
import { media } from "@yaebal/core";
import { files } from "@yaebal/files";
import { createPool } from "@yaebal/workers";
import type { Tasks } from "./tasks.js";

const pool = createPool<Tasks>(new URL("./tasks.js", import.meta.url), { size: 4 });

bot.install(files()).on("message:photo", async (ctx) => {
  const largest = ctx.message.photo.at(-1)!;
  const bytes = await ctx.files.download(largest.file_id);
  const thumb = await pool.run("resize", bytes, { timeout: 5_000 }); // typed: Uint8Array in, Uint8Array out
  await ctx.sendPhoto(media.buffer(thumb));
});

// on shutdown — drain in-flight tasks, then terminate
await pool.close();

bot integration

@yaebal/workers/plugin puts the pool on the context as ctx.tasks and closes it when the bot stops, so you never leak a pool on shutdown. install it on the Bot itself so the plugin can hook onStop. drivers that own their own loop (like @yaebal/runner) don't fire the bot's stop handlers — install with tasks(pool, { onStop: false }) and close the pool yourself.

plugin.ts

// wire the pool onto the context and tie its lifecycle to the bot
import { tasks } from "@yaebal/workers/plugin";

const bot = new Bot(token).install(tasks(pool)); // ctx.tasks, closed on bot.stop()

bot.command("thumb", async (ctx) => {
  const bytes = await ctx.files.download(fileId);
  await ctx.sendPhoto(media.buffer(await ctx.tasks.run("resize", bytes)));
});

timeouts & cancellation

every task handler gets an AbortSignal as its second argument. a timeout (per call or pool-wide) or an external signal aborts it: settle promptly when the signal fires and the worker survives; ignore it and the worker is terminated after killTimeout (default 500ms) and respawned. a queued task that's aborted is dropped before it ever runs.

cancel.ts

// per call: a timeout (or an external AbortSignal) cancels the task
const controller = new AbortController();
const scored = pool.run("score", input, { timeout: 3_000, signal: controller.signal });

// worker side: settle promptly when the signal fires and the worker keeps living
register<Tasks>({
  score: ({ text, rounds }, { signal }) => {
    for (let i = 0; i < rounds; i++) {
      if (signal.aborted) throw new Error("aborted");
      // …crunch…
    }
    return result;
  },
});

transferables & zero-copy

pass transfer to move a buffer to the worker instead of copying it, and wrap a result in move() to move it back — the bytes change owners without a clone.

transfer.ts

// move the buffer to the worker (no copy), and move the result back
import { move } from "@yaebal/workers";

register<Tasks>({ resize: (buf) => move(out, [out.buffer]) }); // ← move back
const out = await pool.run("resize", bytes, { transfer: [bytes.buffer] }); // ← move in

resilience

a worker that crashes, is killed after a timeout, or trips its resourceLimits rejects only its in-flight tasks and respawns with backoff; a worker file that can't even start is retried up to maxRestarts times, then the slot is declared dead — no fork-bomb, no infinite respawn loop. maxQueue bounds the backlog, pool.ready() surfaces startup failures, and pool.stats() / pool.on(…) give you observability.

resilience.ts

// crash recovery, backpressure and observability come built in
const pool = createPool<Tasks>(workerFile, {
  size: "auto",              // availableParallelism() - 1
  maxQueue: 512,             // run() rejects with QueueFullError past this
  worker: { resourceLimits: { maxOldGenerationSizeMb: 256 } }, // a bomb crashes one worker, alone
});

await pool.ready();          // resolves when every worker registered, rejects if it can't start
pool.on("worker:crash", ({ worker, code, willRespawn }) =>
  console.warn("worker", worker, "died", code, willRespawn ? "(respawning)" : "(dead)"));

pool.stats(); // { size, ready, busy, dead, queued, running, completed, failed, restarts }

api

export	signature	description
`createPool`	`<Tasks>(workerFile: string \| URL, options?: PoolOptions) => Pool<Tasks>`	spawn a pool of workers
`register`	`<Tasks>(handlers: TaskHandlers<Tasks>) => void`	called inside the worker file to expose tasks
`move`	`<T>(value: T, transfer: Transferable[]) => T`	move a task result back to the main thread, no copy
`tasks`	`(pool, options?) => Plugin`	`/plugin` — expose the pool as `ctx.tasks`
`isWorkerThread`	`boolean`	guard `register()` in single-file setups

PoolOptions

option	type	default	description
`size`	`number \| "auto"`	`1`	worker threads; `"auto"` = `availableParallelism() - 1`
`concurrency`	`number`	`1`	tasks a single worker runs at once
`maxQueue`	`number`	`Infinity`	max tasks waiting before `run` throws `QueueFullError`
`timeout`	`number` (ms)	none	default per-task execution timeout
`killTimeout`	`number` (ms)	`500`	grace period for an aborted task before its worker is killed
`maxRestarts`	`number`	`5`	consecutive crashes-without-ready before a slot is declared dead
`worker`	`PoolWorkerOptions`	—	`workerData` / `resourceLimits` / `env` / `execArgv`

Pool

member	signature	description
`run`	`(name, arg?, options?: RunOptions) => Promise<result>`	run a task on the next free worker; `arg` and result inferred from `Tasks[name]`
`ready`	`() => Promise<void>`	resolves when all workers registered; rejects if the pool can't start
`close`	`(options?: { timeout?: number }) => Promise<void>`	drain queued + running tasks, then terminate
`destroy`	`() => Promise<void>`	terminate now; reject everything in flight
`stats`	`() => PoolStats`	queue depth, worker states, lifetime counters
`on`	`(event, listener) => () => void`	`"worker:ready"` / `"worker:crash"`; returns unsubscribe
`size`	`number`	number of worker threads (readonly)

RunOptions is { transfer?, signal?, timeout? }. errors thrown in a task cross the thread with their name, message and stack intact; pool errors are typed — QueueFullError, TaskTimeoutError, UnknownTaskError, WorkerCrashError, PoolClosedError (all extend PoolError).

built .js only. the worker file must be a built .js (or run under a ts loader — workers inherit the parent's --experimental-strip-types). workers don't share closures with the main thread; they only receive the data you pass to run. register() throws outside a worker thread — guard with isWorkerThread in single-file setups. Pool implements Symbol.asyncDispose, so await using pool = createPool(…) destroys it at scope exit.