Skip to content

API.md

Latest commit

 

History

History
649 lines (489 loc) · 12.9 KB

API.md

File metadata and controls

649 lines (489 loc) · 12.9 KB
Raw

API Reference

This is the public API for billiards: pool creation, task scheduling, keyed concurrency, retries, cancellation, backpressure, runtime resizing, stats, and exported errors.

Import

import { createPool } from "billiards"

CommonJS is also supported:

const { createPool } = require("billiards")

The package exposes ESM, CommonJS, and TypeScript declaration entries.

Create A Pool

const pool = createPool({ concurrency: 8 })

createPool(options)

declare function createPool(options: PoolOptions): Pool

PoolOptions

type PoolOptions = {
  concurrency: number
  perKeyConcurrency?: number
  timeout?: number
  retry?: RetryPolicy | false
  hooks?: PoolHooks
}

concurrency

Maximum number of tasks that can run at once globally.

Must be a positive integer.

perKeyConcurrency

Maximum number of tasks that can run at once for the same key.

Behavior:

  • if unset and a task has a key, defaults to 1
  • if set, must be a positive integer
  • tasks without keys are limited only by global concurrency

timeout

Optional timeout in milliseconds for each task attempt.

Behavior:

  • timeout starts when an attempt begins, not when a task enters the queue
  • timeout aborts the task context signal
  • timeout produces a PoolTimeoutError
  • retry policy may retry timeout errors unless retryIf says otherwise

retry

Optional retry behavior. Retries are disabled by default.

type RetryPolicy = {
  attempts: number
  backoff?: "fixed" | "exponential" | RetryBackoff
  minDelay?: number
  maxDelay?: number
  jitter?: boolean | number
  retryIf?: RetryPredicate
}

attempts is the total number of attempts, including the first attempt. minDelay defaults to 0, so retries are immediate unless a delay is configured.

const pool = createPool({
  concurrency: 8,
  retry: { attempts: 3 }
})

This means: try once, then retry up to two more times.

type RetryBackoff = (ctx: RetryContext) => number

type RetryPredicate = (
  error: unknown,
  ctx: RetryContext
) => boolean | Promise<boolean>

type RetryContext = {
  attempt: number
  attempts: number
  key?: TaskKey
  delay: number
}

hooks

Optional lifecycle hooks for lightweight observability.

type PoolHooks = {
  onStart?: (event: TaskStartEvent) => void
  onSuccess?: (event: TaskSuccessEvent) => void
  onError?: (event: TaskErrorEvent) => void
  onRetry?: (event: TaskRetryEvent) => void
  onCancel?: (event: TaskCancelEvent) => void
}

Hooks are for observability, not scheduling control. If a hook throws, the pool ignores the hook error so observability cannot break task execution.

Task Function

type Task<T> = (ctx: TaskContext) => T | Promise<T>
type TaskContext = {
  signal: AbortSignal
  attempt: number
  key?: TaskKey
}

attempt is one-based. The first attempt is 1.

Task Keys

type TaskKey = string | number | symbol

Keys are used for per-key concurrency.

await pool.run(task, { key: customerId })

For map, the key is usually derived from the item:

await pool.map(jobs, processJob, {
  key: job => job.customerId
})

pool.run(task, options?)

Run a task and return its result promise.

const value = await pool.run(async ({ signal }) => {
  return fetchSomething({ signal })
})

Signature:

run<T>(task: Task<T>, options?: TaskOptions): Promise<T>

Use run when you only need the result promise.

pool.add(task, options?)

Add a task and return a handle.

const handle = pool.add(() => uploadFile(file), {
  key: file.accountId
})

const result = await handle.result

Signature:

add<T>(task: Task<T>, options?: TaskOptions): TaskHandle<T>

TaskOptions

type TaskOptions = {
  key?: TaskKey
  signal?: AbortSignal
  timeout?: number
  retry?: RetryPolicy | false
  id?: string
}

Task-level options override pool-level options.

TaskHandle

type TaskHandle<T> = {
  id: string
  key?: TaskKey
  result: Promise<T>
  status: () => TaskStatus
  cancel: (reason?: unknown) => void
}
type TaskStatus =
  | "queued"
  | "running"
  | "retrying"
  | "succeeded"
  | "failed"
  | "cancelled"

pool.map(items, worker, options?)

Map over an iterable with pool scheduling.

const results = await pool.map(urls, async (url, { signal }) => {
  const response = await fetch(url, { signal })
  return response.text()
})

Signature:

map<Input, Output>(
  items: Iterable<Input> | AsyncIterable<Input>,
  worker: MapWorker<Input, Output>,
  options?: MapOptions<Input>
): Promise<Output[]>
type MapWorker<Input, Output> = (
  item: Input,
  ctx: MapTaskContext<Input>
) => Output | Promise<Output>
type MapTaskContext<Input> = TaskContext & {
  index: number
  item: Input
}

Default Behavior

  • output order matches input order
  • fail-fast on first error
  • queued tasks from that map call are cancelled on fail-fast
  • already-running tasks receive aborted signals if cancellation is possible
  • cancellation can reject the map() promise before ignored-abort running workers finish
  • input is pulled lazily rather than fully enqueued up front

pool.mapSettled(items, worker, options?)

Map over items and return every result, whether fulfilled, rejected, or cancelled.

const results = await pool.mapSettled(jobs, processJob)

Signature:

mapSettled<Input, Output>(
  items: Iterable<Input> | AsyncIterable<Input>,
  worker: MapWorker<Input, Output>,
  options?: MapOptions<Input>
): Promise<Array<PoolSettledResult<Output>>>
type PoolSettledResult<T> =
  | { status: "fulfilled"; value: T }
  | { status: "rejected"; reason: unknown }
  | { status: "cancelled"; reason?: unknown }

mapSettled() waits for every started worker to settle before returning its result array. If a running worker ignores an aborted signal forever, mapSettled() cannot produce a final settled record for that item and can remain pending.

MapOptions<Input>

type MapOptions<Input> = {
  key?: (item: Input, index: number) => TaskKey | undefined
  signal?: AbortSignal
  timeout?: number
  retry?: RetryPolicy | false
  buffer?: number
}

buffer

Maximum number of tasks owned by a collection call that may be scheduled ahead of the consumer. For each(), it also bounds completed results waiting to be yielded.

Default: the pool's current concurrency.

This keeps large or async inputs from being eagerly materialized into one pending promise per item.

pool.each(items, worker, options?)

Return an async iterator that yields results as tasks complete.

for await (const result of pool.each(urls, fetchUrl, { buffer: 32 })) {
  console.log(result.index, result.value)
}

Signature:

each<Input, Output>(
  items: Iterable<Input> | AsyncIterable<Input>,
  worker: MapWorker<Input, Output>,
  options: EachOptions<Input> & { settled: true }
): AsyncIterable<EachSettledResult<Input, Output>>

each<Input, Output>(
  items: Iterable<Input> | AsyncIterable<Input>,
  worker: MapWorker<Input, Output>,
  options?: EachOptions<Input> & { settled?: false | undefined }
): AsyncIterable<EachResult<Input, Output>>
type EachOptions<Input> = MapOptions<Input> & {
  settled?: boolean
}
type EachResult<Input, Output> = {
  index: number
  item: Input
  value: Output
}

type EachSettledResult<Input, Output> =
  | ({ status: "fulfilled" } & EachResult<Input, Output>)
  | { status: "rejected"; index: number; item: Input; reason: unknown }
  | { status: "cancelled"; index: number; item: Input; reason?: unknown }

Behavior:

  • yields in completion order
  • defaults to fail-fast on error
  • supports settled: true to yield rejected/cancelled results instead of throwing
  • supports buffer to limit completed-but-not-consumed results
  • if the consumer breaks early, queued tasks owned by that iterator are cancelled and running task signals are aborted
  • default fail-fast each() can reject promptly on cancellation, but each(..., { settled: true }) waits for started workers to settle before yielding their final records

pool.drain()

Resolve when the pool has no pending work: no queued tasks, no running tasks, and no tasks waiting in retry backoff.

await pool.drain()

Signature:

drain(options?: DrainOptions): Promise<void>
type DrainOptions = {
  signal?: AbortSignal
}

drain should not cancel work. It only waits, including for retry-delayed tasks that are still part of the pool.

pool.clearQueue(reason?)

Cancel queued tasks without aborting running tasks.

pool.clearQueue(new Error("shutdown"))
await pool.drain()

Signature:

clearQueue(reason?: unknown): number

Returns the number of queued tasks cancelled.

Queued task result promises should reject with PoolCancelledError. Running tasks are not affected.

pool.cancel(reason?)

Cancel all queued tasks and abort all running task signals.

pool.cancel(new Error("shutdown"))

Signature:

cancel(reason?: unknown): void

Behavior:

  • queued tasks become cancelled immediately
  • running task contexts are aborted
  • running tasks settle when their user code returns or throws
  • drain() resolves only after running tasks finish settling

Collection surfaces expose two cancellation shapes. Fail-fast map() and default each() can reject promptly after cancellation is observed. Settled collection surfaces, mapSettled() and each(..., { settled: true }), wait for actual settled records from running workers; if a worker ignores its aborted signal forever, those settled calls can remain pending.

pool.waitForSizeBelow(limit, options?)

Resolve when the number of queued tasks is below limit.

for await (const job of jobs) {
  await pool.waitForSizeBelow(1_000)
  pool.add(() => processJob(job))
}

Signature:

waitForSizeBelow(limit: number, options?: { signal?: AbortSignal }): Promise<void>

limit must be a positive integer. This is a producer backpressure helper for queued work. It does not cancel or start work by itself.

pool.resize(options)

Change concurrency limits at runtime.

pool.resize({ concurrency: 4 })
pool.resize({ concurrency: 50, perKeyConcurrency: 2 })

Signature:

resize(options: {
  concurrency?: number
  perKeyConcurrency?: number
}): void

Increasing concurrency may start more queued work. Decreasing concurrency should not abort running tasks; it only affects future scheduling.

pool.stats()

Return a snapshot of pool state.

const stats = pool.stats()

Signature:

stats(): PoolStats
type PoolStats = {
  queued: number
  running: number
  completed: number
  succeeded: number
  failed: number
  cancelled: number
  retried: number
  retrying: number
  concurrency: number
  perKeyConcurrency: number
  keys: Array<{
    key: TaskKey
    queued: number
    running: number
    retrying: number
  }>
}

Stats are snapshots. They do not update live.

Errors

Exported errors:

class PoolError extends Error {
  readonly cause?: unknown
}
class PoolTimeoutError extends PoolError {
  readonly timeout: number
  readonly attempt: number
  readonly taskId: string
  readonly key?: TaskKey
}
class PoolCancelledError extends PoolError {
  readonly reason?: unknown
}
class PoolValidationError extends PoolError {}

Recommended error behavior:

  • invalid options throw PoolValidationError
  • timeout throws PoolTimeoutError
  • explicit cancellation throws PoolCancelledError
  • map fail-fast rejects with the first terminal error
  • mapSettled does not throw for task failures

Export Surface

export {
  createPool,
  PoolError,
  PoolTimeoutError,
  PoolCancelledError,
  PoolValidationError
}

export type {
  DrainOptions,
  EachOptions,
  EachResult,
  EachSettledResult,
  MapOptions,
  MapTaskContext,
  MapWorker,
  Pool,
  PoolHooks,
  PoolOptions,
  PoolSettledResult,
  PoolStats,
  RetryBackoff,
  RetryContext,
  RetryPolicy,
  RetryPredicate,
  Task,
  TaskCancelEvent,
  TaskContext,
  TaskErrorEvent,
  TaskHandle,
  TaskKey,
  TaskOptions,
  TaskRetryEvent,
  TaskStartEvent,
  TaskStatus,
  TaskSuccessEvent
}

API examples

Backfill users safely

const pool = createPool({ concurrency: 20 })

await pool.map(users, async user => {
  await backfillUser(user.id)
})

Avoid per-account races

const pool = createPool({
  concurrency: 100,
  perKeyConcurrency: 1
})

await pool.map(events, handleEvent, {
  key: event => event.accountId
})

Retry flaky network tasks

const pool = createPool({
  concurrency: 10,
  retry: {
    attempts: 3,
    backoff: "exponential",
    minDelay: 250,
    maxDelay: 5_000,
    jitter: true
  }
})

await pool.map(webhooks, sendWebhook)

Cancel on shutdown

const shutdown = new AbortController()

process.once("SIGTERM", () => {
  shutdown.abort()
  pool.cancel(new Error("SIGTERM"))
})

await pool.map(jobs, processJob, {
  signal: shutdown.signal
})