@hecto/async

Composable async tasks with cancellation, retry, and concurrency control.

pnpm add @hecto/async

import { Task } from '@hecto/async'
import { pipe } from '@hecto/fp'
import { map as mapResult } from '@hecto/fp/result'

const fetchUser = pipe(
  Task.tryPromise(
    () => fetch('/api/user').then((response) => response.json()),
    (cause) => ({ type: 'FetchUserFailed' as const, cause }),
  ),
  Task.timeout(5000),
  Task.retry({ attempts: 3, backoff: 'exponential' }),
  Task.map(mapResult((data) => data.user)),
)

const result = await Task.run(fetchUser)

Lazy async iteration is available from the root namespace or the focused subpath:

import { AsyncIter, Task } from '@hecto/async'

const values = AsyncIter.from([3, 1, 2])
const mapped = AsyncIter.mapConcurrent(
  values,
  async (value, _index, signal) => {
    await fetch(`/work/${value}`, { signal })
    return value * 2
  },
  { concurrency: 2 },
)

const ordered = await Task.run(AsyncIter.collect(mapped))
// [6, 2, 4] -- output order always matches source order

What's in the box

• Task: lazy cancellable execution with one value parameter; expected failures are Task<Result<A, E>>
• AsyncIter: lazy sync/async iterable pipelines with end-to-end cancellation
• Combinators: value map/flatMap/tap, plus Result-aware mapError, catchError, and match
• Concurrency: all, allSettled, race, any, parallel, sequential
• Resilience: retry, timeout, fallback
• Flow control: throttle, debounce, rateLimit
• Collection: mapAsync, filterAsync, forEachAsync, reduceAsync, collectAsync

Docs

Workers (Bun)

CPU work runs on a Bun worker thread so the main thread stays responsive. Importing @hecto/async/workers starts no worker; @hecto/async/workers/bun is Bun-only (a bun export condition) and starts workers lazily on the first prepare/submit.

Task file — a named export in its own module; nothing but plain data crosses the boundary. Take WorkerContext as the second parameter when the loop needs cooperative cancellation:

// report.worker.ts
import type { WorkerContext } from '@hecto/async/workers'

export async function calculateReport(input: { values: readonly number[] }, context: WorkerContext) {
  let total = 0
  for (let i = 0; i < input.values.length; i++) {
    total += input.values[i]!
    if ((i & 4095) === 0) await context.checkpoint() // yields so CANCEL is seen, throws if aborted
  }
  return { total }
}

Reference — taskModule is inert; the typeof import is erased so the parent never loads task code:

import { taskModule } from '@hecto/async/workers'

export const calculateReportTask = taskModule<typeof import('./report.worker.js')>(
  new URL('./report.worker.js', import.meta.url),
).task('calculateReport')

Pool:

import { createPool } from '@hecto/async/workers/bun'

const pool = createPool({ size: 2, maxQueue: 32 })
try {
  const { total } = await pool.run(calculateReportTask, { values: [1, 2, 3] })
  const job = pool.submit(calculateReportTask, { values: [4, 5] }, { timeoutMs: 1000 })
  await job.result
  await job.completion
} finally {
  const closed = await pool.close({ graceMs: 1000 })
  // closed.status is 'drained' | 'incomplete'; closed.nativeExit is always 'unknown' on Bun 1.4.
}

Limits — size (default max(1, min(4, cores - 1))), maxQueue 64, maxPendingBytes 32 MiB, maxPayloadBytes 8 MiB, taskTimeoutMs 30 s, cancelGraceMs 100 ms, closeGraceMs 5 s. Input is snapshotted with structuredClone on admission (default copy; pass transfer to move ArrayBuffers). A full queue rejects synchronously with WorkerError failure: 'overloaded'; every infrastructure failure is a guarded WorkerError with a bounded failure discriminant, and cancellation is a WorkerCancelledError recognised by isCancellation. Three consecutive crashes mark the pool unhealthy.

nativeExit: 'unknown' — Bun reports close/terminate acknowledgements only; nothing confirms a thread has physically exited. close() therefore reports drained once every registered task settled and each worker acknowledged STOPPED, and nativeExit: 'unknown' always. Do not build shutdown logic that assumes the thread is gone.

Measured on a 14-core M-series host under foreign load (benchmarks/workers/overhead.mjs, responsiveness.mjs): ~19 µs per job sequentially and ~9 µs pipelined for a trivial task; a pool of 4 beats the main thread from about 0.5 ms of CPU per job; a 5 ms main-thread interval keeps a p99 under 7 ms while one or four 500 ms tasks run, against ~500 ms inline.