diff --git a/.claude/skills/verify/SKILL.md b/.claude/skills/verify/SKILL.md new file mode 100644 index 0000000..1831ac7 --- /dev/null +++ b/.claude/skills/verify/SKILL.md @@ -0,0 +1,29 @@ +--- +name: verify +description: Verify await-parallel-limit at its real surface — the npm package boundary (packed tarball installed into a fresh consumer), never via ./src imports. +--- + +# Verifying await-parallel-limit + +Surface: the package boundary. Pack → install into a throwaway consumer → drive as a dependent would. + +## Recipe + +1. Build + pack: `npx tsc && npm pack --pack-destination ` +2. Fresh consumer: `mkdir /consumer && cd there && npm init -y && npm i /await-parallel-limit-.tgz` +3. Drive CJS: `require('await-parallel-limit').default` and named `{ parallel, settle, map, mapSettled, DEFAULT_CONCURRENCY }`. + +## Flows worth driving + +- Legacy dependent shape (`@paperbits/core`): `await parallel(thunks, 30)`, result discarded. +- Ordering (slow first job must still come back first) and peak-concurrency (active/peak counters in jobs). +- Abort: mid-flight, pre-aborted, **synchronous abort from inside a job during startup** (regression: must reject, not fulfill with holes), abort after completion (no-op), one signal reused across 30 runs (no MaxListeners warning). +- Fail-fast: after first rejection, started-count must not grow (remaining jobs abandoned). +- Garbage: non-array, junk limits (0, -1, 2.5, NaN, string, null → default 5), non-function array element (TypeError rejection), sparse arrays through map (holes → mapper gets undefined). +- TS surface: compile a consumer .ts against the *installed* d.ts. Floor is **TS 3.4** (`readonly I[]` in map/mapSettled signatures) — pin with `npm i --no-save typescript@3.4.5`. + +## Gotchas + +- Node ESM default import works via the `exports` map + `esm/index.mjs` wrapper. If `import parallel from` ever yields an object instead of a function, the exports map or wrapper broke — run `test/boundary/smoke.mjs` against the packed tarball. +- `tsc ... | head` masks the exit code — check `$?` on the tsc command itself, not the pipe. +- Scale check: 100k-item `map` should be O(10ms); if it regresses to O(100ms), per-item thunk allocation crept back in. diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 655ffa0..7335aba 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -10,7 +10,8 @@ jobs: runs-on: ubuntu-latest strategy: matrix: - node-version: [16.x, 18.x, 20.x, 22.x] + # 16.14 is the declared engines floor — tested exactly. + node-version: ['16.14', 16.x, 18.x, 20.x, 22.x, 24.x] steps: - uses: actions/checkout@v4 - name: Use Node.js ${{ matrix.node-version }} @@ -19,3 +20,46 @@ jobs: node-version: ${{ matrix.node-version }} - run: npm install - run: npm test + + fuzz: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-node@v4 + with: + node-version: 22.x + - run: npm install + - name: Differential fuzz (500 seeded scenarios) + run: npm run fuzz -- 500 + + package-boundary: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-node@v4 + with: + node-version: 22.x + - run: npm install + - name: Pack tarball + run: | + npm pack --pack-destination /tmp + echo "TARBALL=$(ls /tmp/await-parallel-limit-*.tgz)" >> "$GITHUB_ENV" + - name: Install into a fresh consumer and run the smoke tests (CJS + ESM) + run: | + mkdir -p /tmp/consumer && cd /tmp/consumer + npm init -y > /dev/null + npm install "$TARBALL" + cp "$GITHUB_WORKSPACE/test/boundary/smoke.cjs" "$GITHUB_WORKSPACE/test/boundary/smoke.mjs" . + node --unhandled-rejections=strict smoke.cjs + node --unhandled-rejections=strict smoke.mjs + - name: Typings floor gate — TypeScript 3.4 compiles the installed d.ts + run: | + cd /tmp/consumer + cp "$GITHUB_WORKSPACE/test/boundary/consumer.ts" . + npm install --no-save typescript@3.4.5 + ./node_modules/.bin/tsc --noEmit --strict --target es2017 --lib es2017 --moduleResolution node consumer.ts + - name: Typings gate — latest TypeScript compiles the installed d.ts + run: | + cd /tmp/consumer + npm install --no-save typescript@latest + ./node_modules/.bin/tsc --noEmit --strict --target es2017 --lib es2017 --moduleResolution node consumer.ts diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml new file mode 100644 index 0000000..3fba30f --- /dev/null +++ b/.github/workflows/publish.yml @@ -0,0 +1,63 @@ +name: Publish + +# Publishes to npm with provenance attestations. Trigger by publishing a +# GitHub Release (tag `vX.Y.Z`), or manually via workflow_dispatch. +# +# One-time setup: add an npm Automation token as the `NPM_TOKEN` repository +# secret (Settings → Secrets and variables → Actions). Provenance requires a +# public repo and the `id-token: write` permission below — npm links the +# published tarball to this exact workflow run and commit. + +on: + release: + types: [published] + workflow_dispatch: + +permissions: + contents: read + id-token: write # required for npm provenance (OIDC) + +jobs: + publish: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-node@v4 + with: + node-version: 22.x + registry-url: 'https://registry.npmjs.org' + + - run: npm install + + - name: Guard — tag matches package.json version + if: github.event_name == 'release' + run: | + PKG_VERSION="$(node -p "require('./package.json').version")" + TAG="${{ github.event.release.tag_name }}" + if [ "v$PKG_VERSION" != "$TAG" ] && [ "$PKG_VERSION" != "$TAG" ]; then + echo "::error::package.json is $PKG_VERSION but the release tag is $TAG" + exit 1 + fi + + - name: Full gate — unit suite + type gate + run: npm test + + - name: Full gate — differential fuzz + run: npm run fuzz -- 500 + + - name: Full gate — package-boundary smoke (CJS + ESM) from the tarball + run: | + npm pack --pack-destination /tmp + TARBALL="$(ls /tmp/await-parallel-limit-*.tgz)" + mkdir -p /tmp/consumer && cd /tmp/consumer + npm init -y > /dev/null + npm install "$TARBALL" + cp "$GITHUB_WORKSPACE/test/boundary/smoke.cjs" "$GITHUB_WORKSPACE/test/boundary/smoke.mjs" . + node --unhandled-rejections=strict smoke.cjs + node --unhandled-rejections=strict smoke.mjs + + - name: Publish with provenance + run: npm publish --provenance --access public + env: + NODE_AUTH_TOKEN: ${{ secrets.NPM_TOKEN }} diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..f1ad7a2 --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,56 @@ +# Agent guide — await-parallel-limit + +Zero-dependency npm package: a concurrency-limited promise pool. ~230 lines of +TypeScript in `src/index.ts`; everything else is tests and packaging. + +## Consumer API + +See `llms.txt` (shipped in the npm tarball) for the complete API, semantics, +recipes, and gotchas in one file. + +## Working on this repo + +- Build: `npm run build` (tsc → `dist/`). `dist/` is git-ignored; `prepare` + builds on install/publish. +- Test: `npm test` — builds, compiles `test/types.test.ts` (exact-type + assertions; a wrong inference is a build failure), then runs the unit suite + under `--unhandled-rejections=strict`. +- Fuzz: `npm run fuzz -- [iterations] [seed]` — seeded differential fuzzer + (`test/fuzz.cjs`); failures print the seed for exact reproduction. +- Verify at the real surface: follow `.claude/skills/verify/SKILL.md` — pack + the tarball, install into a fresh consumer, drive `require`/`import` + `'await-parallel-limit'`. Never verify via `./src` imports. + +## Architecture + +One core, four one-line wrappers: + +- `run(items, invoke, limit, settle, signal)` in `src/index.ts` — spawns up to + `limit` workers pulling from a shared cursor; workers call + `invoke(items[i], i)` directly (no per-item thunk allocation). A single + `stopped` flag unifies fail-fast and abort cancellation. +- `parallel`/`settle` pass the jobs array + a call-the-thunk invoker; + `map`/`mapSettled` pass the user's items + the mapper itself. +- ESM entry is a static wrapper (`esm/index.mjs`) re-exporting the CJS build — + do NOT introduce a second compiled implementation (dual-package hazard). + +## Invariants you must not break (enforced by tests + fuzzer + CI) + +1. Results in input order; peak in-flight ≤ effective limit; sliding pool + (finished task replaced immediately — never batches). +2. Legacy call shape `parallel(jobs, 30)` and the default export are frozen — + published dependents rely on them. +3. Limit normalization: non-positive/non-integer → 5; `Infinity` → unbounded. +4. Fail-fast stops scheduling; settle mode never rejects on job errors. +5. Abort (even synchronous, from inside a job) rejects with `signal.reason`; + listener always removed; no unhandled rejections ever. +6. Zero runtime dependencies; compiles without DOM/es2020 libs; typings floor + TS 3.4 (`readonly` array syntax — nothing newer in the public d.ts). +7. Node floor 16.14 (`engines`), tested exactly in CI. + +## Conventions + +- No new runtime dependencies. Keep declaration-emit clean (JSDoc flows into + `dist/index.d.ts` — it is consumer documentation). +- Every bug fix lands with a regression test; behavioral claims in README must + have a matching CI gate. diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..34c4185 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,12 @@ +# await-parallel-limit + +Follow `AGENTS.md` — it has the build/test/fuzz commands, architecture map, +and the invariants this package must never break. Consumer-facing API +reference lives in `llms.txt`. + +Quick anchors: +- `npm test` = build + exact-type gate + unit suite (strict unhandled rejections) +- `npm run fuzz -- 300` before any change to `src/index.ts` +- Package-boundary verification recipe: `.claude/skills/verify/SKILL.md` +- Never break: input-order results, sliding-pool concurrency, the v2 call shape + `parallel(jobs, limit)`, zero dependencies, TS 3.4 typings floor. diff --git a/README.md b/README.md index 03a414a..10a6ae1 100644 --- a/README.md +++ b/README.md @@ -1,63 +1,131 @@ # await-parallel-limit -Run an array of async functions with a bounded number running concurrently. +Run async work with a bounded number of tasks in flight at once. Zero dependencies, first-class TypeScript types. ```bash npm install await-parallel-limit --save ``` -## Behaviour - -- Runs at most `concurrency` jobs at a time (default `5`). -- Results are returned in **input order**, not completion order. -- If any job rejects, the returned promise rejects with the first error - (same semantics as `Promise.all`). Jobs already in flight still run to - completion, but their results are discarded. -- A `concurrency` value that is not a positive integer (e.g. `0`, `-1`, `2.5`) - falls back to the default of `5`. +## Guarantees + +- **Concurrency ceiling:** at most `concurrency` tasks are in flight at any + moment (default `5`; `Infinity` means unbounded). A `concurrency` that is not + a positive integer or `Infinity` (e.g. `0`, `-1`, `2.5`, `NaN`) falls back to + the default of `5`. +- **Sustained concurrency:** when a task finishes, the next one starts + immediately — a sliding worker pool, not fixed batches. +- **Ordering:** results are returned in **input order**, not completion order. +- **Fail-fast** (`parallel` / `map`): on the first rejection the returned + promise rejects with that error (like `Promise.all`); no further tasks are + started, and in-flight tasks run to completion but their results are + discarded. Use `settle` / `mapSettled` to collect every outcome instead + (like `Promise.allSettled`). +- **Abort** (`{ signal }`): an already-aborted signal rejects before any task + starts; aborting mid-run — even synchronously from inside a task — rejects + with the signal's `reason` and stops scheduling. The abort listener is always + removed, so one long-lived signal can be reused across many runs. +- **No stray rejections:** after the returned promise settles, a still-running + task that later rejects is absorbed — it never surfaces as an unhandled + rejection. +- **Input snapshot:** the input array's length is captured at call time, so + mutating the array mid-run cannot change the result set. Elements are read + lazily at dispatch; don't mutate the input during a run. + +These guarantees are enforced by the unit suite plus a seeded differential +fuzzer (`npm run fuzz`) and a package-boundary smoke test in CI. ## API ```typescript -parallel( - jobs: Array<() => Promise>, - concurrency?: number, // default: 5 -): Promise +type Options = { signal?: AbortSignal } + +// Run an array of thunks, fail-fast, results in input order. +parallel(jobs: Array<() => Promise>, concurrency?: number, options?: Options): Promise + +// Like parallel, but never rejects on a failing job — returns per-job outcomes. +settle(jobs: Array<() => Promise>, concurrency?: number, options?: Options): Promise[]> + +// Map over data with a mapper, fail-fast, results in input order. +map(items: I[], mapper: (item: I, index: number) => R | Promise, concurrency?: number, options?: Options): Promise + +// Like map, but never rejects — returns per-item outcomes. +mapSettled(items: I[], mapper: (item: I, index: number) => R | Promise, concurrency?: number, options?: Options): Promise[]> + +type SettledResult = + | { status: 'fulfilled'; value: T } + | { status: 'rejected'; reason: any } ``` -## TypeScript +All four functions are named exports; `parallel` is also the default export. + +## Examples ```typescript -import parallel from 'await-parallel-limit' -// named import also available: import { parallel } from 'await-parallel-limit' +import parallel, { settle, map, mapSettled } from 'await-parallel-limit' +// 1. Array of thunks (ordered-tuple typing when declared `as const`). const jobs = [ async () => true, async () => 2, ] as const +const results = await parallel(jobs, 2) // const results: [boolean, number] + +// 2. Map over data — no need to pre-build closures. +const bodies = await map(urls, (url) => fetch(url).then((r) => r.text()), 10) + +// 3. Don't fail fast — inspect every outcome. +const outcomes = await mapSettled(urls, (url) => fetch(url), 10) +const failed = outcomes.filter((o) => o.status === 'rejected') -// When `jobs` is a readonly tuple (`as const`), the result is an ordered tuple -// typed from each job's return type: -// const results: [boolean, number] -const results = await parallel(jobs, 2) +// 4. Cancel early with an AbortSignal. +const controller = new AbortController() +setTimeout(() => controller.abort(), 5000) +await parallel(jobs, 5, { signal: controller.signal }) ``` -## JavaScript +## Importing + +Both module systems are first-class (an `exports` map routes each to the right +entry — the ESM default import is the function, not a namespace object): ```javascript +// ESM +import parallel, { settle, map, mapSettled } from 'await-parallel-limit' + +// CommonJS const parallel = require('await-parallel-limit').default +const { settle, map, mapSettled } = require('await-parallel-limit') -const jobs = [ - async () => { /* ... */ }, - async () => { /* ... */ }, +const results = await parallel([ async () => { /* ... */ }, async () => { /* ... */ }, -] - -const results = await parallel(jobs, 2) +], 2) ``` +## For AI agents + +The npm tarball ships an [`llms.txt`](./llms.txt) — the complete API, +semantics, recipes, and gotchas in one compact file +(`node_modules/await-parallel-limit/llms.txt`). Contributors: see +[`AGENTS.md`](./AGENTS.md) for build/test/fuzz commands and the invariants +this package guarantees. + +## Compatibility + +`3.x` keeps the `2.x` API: the default export and the `parallel(jobs, limit)` +call shape are unchanged, and `settle`, `map`, `mapSettled`, and the `options` +argument are additive. + +One deliberate behavioural change: after a fail-fast rejection, `3.x` stops +starting the remaining jobs. In `2.x`, surviving workers kept executing every +remaining job in the background even though the batch promise had already +rejected and the results were discarded. If you relied on those background side +effects, use `settle`, which always runs every job. + +Requires Node >= 16.14. TypeScript consumers need TS >= 3.4 (the published +typings use `readonly` array syntax). + ## License MIT diff --git a/esm/index.d.mts b/esm/index.d.mts new file mode 100644 index 0000000..f7269e8 --- /dev/null +++ b/esm/index.d.mts @@ -0,0 +1,11 @@ +export { + default, + parallel, + settle, + map, + mapSettled, + DEFAULT_CONCURRENCY, + SettledResult, + AbortSignalLike, + ParallelOptions, +} from '../dist/index.js' diff --git a/esm/index.mjs b/esm/index.mjs new file mode 100644 index 0000000..1c23588 --- /dev/null +++ b/esm/index.mjs @@ -0,0 +1,13 @@ +// Static ESM wrapper over the CJS build. Node's ESM loader gives a CJS module's +// `module.exports` as the default import, so without this wrapper +// `import parallel from 'await-parallel-limit'` would yield the exports object +// (Node ignores the `__esModule` marker). Re-exporting the same objects keeps a +// single implementation — no dual-package hazard. +import cjs from '../dist/index.js' + +export default cjs.default +export const parallel = cjs.parallel +export const settle = cjs.settle +export const map = cjs.map +export const mapSettled = cjs.mapSettled +export const DEFAULT_CONCURRENCY = cjs.DEFAULT_CONCURRENCY diff --git a/llms.txt b/llms.txt new file mode 100644 index 0000000..abada46 --- /dev/null +++ b/llms.txt @@ -0,0 +1,77 @@ +# await-parallel-limit + +> Zero-dependency promise pool: run async functions with a concurrency limit. +> Ordered results, settle-all mode, map helper, AbortSignal cancellation. +> CJS + ESM. Node >= 16.14. TypeScript >= 3.4 (typings), tuple inference for +> `as const` job arrays. + +## Install + +npm install await-parallel-limit + +## Import (both work everywhere; ESM default import is safe) + +CJS: const parallel = require('await-parallel-limit').default + const { parallel, settle, map, mapSettled, DEFAULT_CONCURRENCY } = require('await-parallel-limit') +ESM: import parallel, { settle, map, mapSettled, DEFAULT_CONCURRENCY } from 'await-parallel-limit' + +## API (all four share the same limit + options semantics) + +parallel(jobs: Array<() => Promise>, limit?: number, options?: { signal?: AbortSignal }): Promise + Run zero-arg async functions. Fail-fast (first rejection rejects the promise, + like Promise.all). Results in INPUT order. With `as const` tuple jobs, the + result type is an ordered tuple. Also the default export. + +settle(jobs, limit?, options?): Promise[]> + Like parallel but never rejects because of a failing job. Each slot is + { status: 'fulfilled', value } or { status: 'rejected', reason } + (concurrency-limited Promise.allSettled). Still rejects on abort. + +map(items: I[], mapper: (item: I, index: number) => R | Promise, limit?, options?): Promise + Map data through an async (or sync) mapper with bounded concurrency. + Fail-fast, input order. Sparse holes reach the mapper as undefined. + +mapSettled(items, mapper, limit?, options?): Promise[]> + Settle-all variant of map. + +DEFAULT_CONCURRENCY = 5 + +## Semantics (guaranteed, fuzz-tested) + +- At most `limit` tasks in flight; a finished task is replaced immediately + (sliding pool, not batches). Infinity = unbounded. +- limit that is not a positive integer or Infinity (0, -1, 2.5, NaN, undefined) + falls back to 5. Non-array input rejects with "First argument is not an array." +- Fail-fast: after the first rejection NO further tasks start; in-flight tasks + finish but results are discarded. Need every task to run? Use settle/mapSettled. +- Abort: pre-aborted signal rejects before anything starts; abort mid-run (even + synchronously from inside a task) rejects with signal.reason and stops + scheduling. Listener always removed — safe to reuse one signal across runs. +- After the returned promise settles, a late task rejection is absorbed (never + an unhandled rejection). +- Input array length is snapshotted at call time; mutating it mid-run does not + change the result set. + +## Recipes + +// Rate-limited fetches, tolerate failures, know which failed: +const outcomes = await mapSettled(urls, (u) => fetch(u), 10) +const failed = urls.filter((_, i) => outcomes[i].status === 'rejected') + +// Typed tuple from heterogeneous jobs: +const jobs = [async () => 1, async () => 'a'] as const +const [n, s] = await parallel(jobs, 2) // n: number, s: string + +// Cancellation with timeout: +const c = new AbortController() +const t = setTimeout(() => c.abort(new Error('timeout')), 5000) +try { return await map(items, work, 8, { signal: c.signal }) } +finally { clearTimeout(t) } + +## Gotchas + +- parallel/settle take THUNKS (() => Promise), not promises. An array of + already-created promises is already running — wrap creation in functions, + or use map(items, mapper) and let the library call you. +- settle/mapSettled reject ONLY on abort or non-array input, never on job errors. +- Jobs cannot be force-cancelled mid-await; abort stops SCHEDULING. diff --git a/package.json b/package.json index 4643452..a206136 100644 --- a/package.json +++ b/package.json @@ -1,18 +1,34 @@ { "name": "await-parallel-limit", - "version": "2.2.0", - "description": "Await an array of async functions with parallel limit.", + "version": "3.0.0", + "description": "Run async functions with a concurrency limit. Zero-dependency promise pool with ordered results, settle-all mode, a map helper, and AbortSignal cancellation. TypeScript tuple inference included.", "main": "dist/index.js", "types": "dist/index.d.ts", + "exports": { + ".": { + "import": { + "types": "./esm/index.d.mts", + "default": "./esm/index.mjs" + }, + "require": { + "types": "./dist/index.d.ts", + "default": "./dist/index.js" + } + }, + "./package.json": "./package.json" + }, "files": [ - "dist" + "dist", + "esm", + "llms.txt" ], "sideEffects": false, "scripts": { "build": "tsc", "clean": "rm -rf dist", "prepare": "tsc", - "test": "tsc && node test/index.test.cjs" + "test": "tsc && tsc --noEmit --strict --target es2017 --lib es2017 --moduleResolution node test/types.test.ts && node --unhandled-rejections=strict test/index.test.cjs", + "fuzz": "tsc && node --unhandled-rejections=strict test/fuzz.cjs" }, "keywords": [ "async", @@ -20,14 +36,28 @@ "array", "parallel", "concurrency", + "concurrency-limit", "limit", "promise", - "pool" + "promise-pool", + "pool", + "p-limit", + "p-map", + "allsettled", + "settle", + "batch", + "rate-limit", + "throttle", + "abort", + "abortsignal", + "cancel", + "typescript", + "zero-dependency" ], "author": "Ondrej Machek, acrocz@gmail.com", "license": "MIT", "engines": { - "node": ">=12" + "node": ">=16.14" }, "repository": { "type": "git", diff --git a/src/index.ts b/src/index.ts index 5db6b9f..a7ac681 100644 --- a/src/index.ts +++ b/src/index.ts @@ -5,57 +5,229 @@ type Unpacked = type MapToResult = { [K in keyof T]: Unpacked> } +/** Per-item outcome for the settle-all variants. Mirrors `PromiseSettledResult` + * but is defined locally so the package needs no `es2020`/DOM lib. */ +export type SettledResult = + | { status: 'fulfilled'; value: T } + | { status: 'rejected'; reason: any } + +type MapToSettled = { [K in keyof T]: SettledResult>> } + +/** Minimal structural shape of an `AbortSignal`. Declared locally so consumers + * don't need the DOM lib; a real `AbortController().signal` satisfies it. + * `reason` is `any` (not `unknown`) so this interface adds no TypeScript-version + * constraint of its own; the package's effective typings floor is TS 3.4, set by + * the `readonly` array syntax in the `map`/`mapSettled` signatures. */ +export interface AbortSignalLike { + readonly aborted: boolean + readonly reason?: any + addEventListener(type: 'abort', listener: () => void, options?: { once?: boolean }): void + removeEventListener(type: 'abort', listener: () => void): void +} + +export interface ParallelOptions { + /** + * Cancel the run early. When the signal aborts, the returned promise rejects + * with the signal's `reason` and no further jobs are started. Jobs already in + * flight are not (and cannot be) force-cancelled — they run to completion but + * their results are discarded. + */ + signal?: AbortSignalLike +} + export const DEFAULT_CONCURRENCY = 5 +// `Infinity` means "no limit" (run everything at once) — 2.x behaviour. +const normalizeLimit = (limit: number | undefined): number => + limit === Infinity ? Infinity : + typeof limit === 'number' && Number.isInteger(limit) && limit > 0 ? limit : DEFAULT_CONCURRENCY + +const abortReason = (signal: AbortSignalLike): unknown => { + if (signal.reason !== undefined) return signal.reason + const err = new Error('The operation was aborted') + err.name = 'AbortError' + return err +} + /** - * Run an array of async, zero-argument job functions with a bounded number of - * jobs in flight at any one time. + * Shared worker-pool core for all four public variants. Spawns up to + * `concurrency` long-lived workers that pull from a shared cursor, so a worker + * that finishes an item immediately picks up the next unclaimed one (sustained + * concurrency, not batching). Results are written back at each item's index, so + * ordering matches the input regardless of completion order. * - * Results are returned in the same order as `jobs` (not completion order). When - * `jobs` is a `readonly` tuple (e.g. declared `as const`), the result type is an - * ordered tuple matching each job's resolved value. - * - * If any job rejects, the returned promise rejects with the first such error - * (matching `Promise.all` semantics); jobs already in flight still run to - * completion but their results are discarded. - * - * @param jobs Array of functions, each returning a promise. - * @param limit Maximum number of jobs to run concurrently. Values that are not - * positive integers fall back to {@link DEFAULT_CONCURRENCY} (5). + * Workers call `invoke(items[i], i)` directly instead of materialising a thunk + * per element: `parallel`/`settle` pass the jobs array with a call-the-thunk + * invoker, `map`/`mapSettled` pass the user's items with the mapper itself. + * This avoids a closure + an extra promise per item on large inputs. */ -const parallel = async ( - jobs: { [K in keyof T]: () => Promise }, - limit: number = DEFAULT_CONCURRENCY, -): Promise> => { - if (!Array.isArray(jobs)) { +const run = async ( + items: readonly any[], + invoke: (item: any, index: number) => any, + limit: number | undefined, + settle: boolean, + signal: AbortSignalLike | undefined, +): Promise => { + if (!Array.isArray(items)) { throw new Error('First argument is not an array.') } + if (signal && signal.aborted) { + throw abortReason(signal) + } - // Normalise the concurrency limit. Non-integer / non-positive values fall back - // to the default — this preserves the historical behaviour where a falsy limit - // (e.g. `0` or `undefined`) meant "use the default of 5". - const concurrency = Number.isInteger(limit) && limit > 0 ? limit : DEFAULT_CONCURRENCY - - const results: any = new Array(jobs.length) + const concurrency = normalizeLimit(limit) + // Snapshot the length so mutating `items` mid-run cannot silently shrink or + // grow the result set; elements themselves are still read lazily at dispatch. + const total = items.length + const results: any[] = new Array(total) let index = 0 + // Single stop mechanism: set on the first fail-fast rejection and by the + // abort listener, so once the returned promise settles early, workers stop + // pulling new items. Items already in flight run to completion but their + // results are discarded by the caller. + let stopped = false const worker = async (): Promise => { while (true) { + if (stopped) return const i = index++ - if (i >= jobs.length) return - results[i] = await jobs[i]() + if (i >= total) return + if (settle) { + try { + results[i] = { status: 'fulfilled', value: await invoke(items[i], i) } + } catch (reason) { + results[i] = { status: 'rejected', reason } + } + } else { + try { + results[i] = await invoke(items[i], i) + } catch (err) { + stopped = true + throw err + } + } } } - const workerCount = Math.min(concurrency, jobs.length) + const workerCount = Math.min(concurrency, total) const workers: Promise[] = [] for (let w = 0; w < workerCount; w++) { workers.push(worker()) } - await Promise.all(workers) + const all = Promise.all(workers) + + if (!signal) { + await all + return results + } - return results as MapToResult + let onAbort: (() => void) | undefined + try { + return await new Promise((resolve, reject) => { + onAbort = () => { + stopped = true + reject(abortReason(signal)) + } + signal.addEventListener('abort', onAbort, { once: true }) + // A job can abort the signal synchronously while the workers are still + // starting up — before the listener above exists. Real signals do not + // fire 'abort' for listeners added after the fact, so re-check by hand. + if (signal.aborted) onAbort() + // `reject` here also absorbs a straggler's rejection arriving after an + // abort has settled this promise, so it never becomes unhandled. + all.then(() => resolve(results), reject) + }) + } finally { + if (onAbort) signal.removeEventListener('abort', onAbort) + } } +const callThunk = (job: () => Promise) => job() + +/** + * Run an array of async, zero-argument job functions with a bounded number of + * jobs in flight at any one time. + * + * Results are returned in the same order as `jobs` (not completion order). When + * `jobs` is a `readonly` tuple (e.g. declared `as const`), the result type is an + * ordered tuple matching each job's resolved value. + * + * If any job rejects, the returned promise rejects with the first such error + * (matching `Promise.all` semantics); no further jobs are started, and jobs + * already in flight run to completion but their results are discarded. Use + * {@link settle} to collect every outcome instead of failing fast. + * + * @param jobs Array of functions, each returning a promise. + * @param limit Max jobs to run concurrently (`Infinity` = unbounded). Values + * that are not positive integers fall back to + * {@link DEFAULT_CONCURRENCY} (5). + * @param options Optional `{ signal }` to cancel the run early. + * @example + * const jobs = [async () => 1, async () => 'a'] as const + * const [n, s] = await parallel(jobs, 2) // n: number, s: string + */ +const parallel = ( + jobs: { [K in keyof T]: () => Promise }, + limit?: number, + options?: ParallelOptions, +): Promise> => + run(jobs as readonly any[], callThunk, limit, false, options && options.signal) as Promise> + +/** + * Like {@link parallel}, but never rejects because of a failing job. Resolves to + * an array of per-job outcomes (`{ status: 'fulfilled', value }` or + * `{ status: 'rejected', reason }`) in input order — the concurrency-limited + * equivalent of `Promise.allSettled`. + * + * @example + * const outcomes = await settle(jobs, 5) + * const failed = outcomes.filter((o) => o.status === 'rejected') + */ +const settle = ( + jobs: { [K in keyof T]: () => Promise }, + limit?: number, + options?: ParallelOptions, +): Promise> => + run(jobs as readonly any[], callThunk, limit, true, options && options.signal) as Promise> + +/** + * Map over `items` with a concurrency limit, calling `mapper(item, index)` for + * each and resolving to the results in input order. The convenience form of + * {@link parallel} when you have data plus a transform rather than pre-built + * thunks. Fails fast on the first rejection. Sparse-array holes are passed to + * the mapper as `undefined` (matching `Promise.all`). + * + * @param items Array of inputs. + * @param mapper `(item, index) => value | Promise`. + * @param limit Max concurrent calls (default {@link DEFAULT_CONCURRENCY}). + * @param options Optional `{ signal }` to cancel the run early. + * @example + * const bodies = await map(urls, (url) => fetch(url).then((r) => r.text()), 10) + */ +const map = ( + items: readonly I[], + mapper: (item: I, index: number) => R | Promise, + limit?: number, + options?: ParallelOptions, +): Promise => + run(items, mapper, limit, false, options && options.signal) as Promise + +/** + * Like {@link map}, but never rejects because of a failing mapper call. Resolves + * to an array of per-item outcomes in input order — the concurrency-limited + * equivalent of `Promise.allSettled` over a mapped array. + * + * @example + * const outcomes = await mapSettled(urls, (url) => fetch(url), 10) + * const failedUrls = urls.filter((_, i) => outcomes[i].status === 'rejected') + */ +const mapSettled = ( + items: readonly I[], + mapper: (item: I, index: number) => R | Promise, + limit?: number, + options?: ParallelOptions, +): Promise[]> => + run(items, mapper, limit, true, options && options.signal) as Promise[]> + export default parallel -export { parallel } +export { parallel, settle, map, mapSettled } diff --git a/test/boundary/consumer.ts b/test/boundary/consumer.ts new file mode 100644 index 0000000..9e339f0 --- /dev/null +++ b/test/boundary/consumer.ts @@ -0,0 +1,30 @@ +/** + * Package-boundary TypeScript consumer. CI compiles this against the INSTALLED + * package's d.ts with both the floor compiler (TypeScript 3.4) and the latest, + * so a typings change that raises the floor or breaks inference fails the build. + * Kept to TS 3.4-compatible syntax on purpose. + */ +import parallel, { settle, map, mapSettled, SettledResult } from 'await-parallel-limit' + +const jobs = [ + async () => true, + async () => 2, +] as const + +async function main(): Promise { + const r = await parallel(jobs, 2) + const a: boolean = r[0] + const b: number = r[1] + + const s = await settle(jobs) + if (s[0].status === 'fulfilled') { + const x: boolean = s[0].value + } + + const m: string[] = await map([1, 2], (n) => `${n}`, 2) + const ms: SettledResult[] = await mapSettled([1, 2], async (n) => n) + + void [a, b, m, ms] +} + +void main diff --git a/test/boundary/smoke.cjs b/test/boundary/smoke.cjs new file mode 100644 index 0000000..4e3c2da --- /dev/null +++ b/test/boundary/smoke.cjs @@ -0,0 +1,81 @@ +'use strict' + +/** + * Package-boundary smoke test. Runs against the INSTALLED package + * (`require('await-parallel-limit')`), not ../src or ../dist — CI packs the + * tarball, installs it into a scratch consumer, copies this file in, and runs + * it. Covers the flows a dependent actually exercises. + */ +const assert = require('assert') +const parallel = require('await-parallel-limit').default +const { parallel: named, settle, map, mapSettled, DEFAULT_CONCURRENCY } = require('await-parallel-limit') + +const delay = (ms) => new Promise((r) => setTimeout(r, ms)) + +;(async () => { + // Legacy dependent shape (@paperbits/core): thunks + integer limit, result discarded. + let done = 0 + const tasks = [] + for (let i = 0; i < 12; i++) tasks.push(() => delay(3).then(() => { done++ })) + await parallel(tasks, 30) + assert.strictEqual(done, 12) + + // Exports. + assert.strictEqual(named, parallel) + assert.strictEqual(DEFAULT_CONCURRENCY, 5) + + // Ordering + concurrency ceiling. + let active = 0, peak = 0 + const jobs = Array.from({ length: 20 }, (_, i) => async () => { + active++; peak = Math.max(peak, active) + await delay(i === 0 ? 25 : 4) + active-- + return i + }) + const ordered = await parallel(jobs, 4) + assert.strictEqual(peak, 4) + assert.deepStrictEqual(ordered, Array.from({ length: 20 }, (_, i) => i)) + + // settle / map / mapSettled. + const s = await settle([async () => 'ok', async () => { throw new Error('bad') }], 2) + assert.deepStrictEqual(s.map((r) => r.status), ['fulfilled', 'rejected']) + assert.deepStrictEqual(await map([1, 2, 3], async (n, i) => `${i}:${n * 2}`, 2), ['0:2', '1:4', '2:6']) + const ms = await mapSettled([1, 2], async (n) => { if (n === 2) throw new Error('x'); return n }, 2) + assert.deepStrictEqual(ms.map((r) => r.status), ['fulfilled', 'rejected']) + + // Abort: mid-flight and synchronous-from-a-job. + const c1 = new AbortController() + const p1 = parallel(Array.from({ length: 10 }, () => () => delay(20)), 2, { signal: c1.signal }) + await delay(5) + c1.abort(new Error('cancelled')) + await assert.rejects(() => p1, /cancelled/) + + const c2 = new AbortController() + await assert.rejects( + () => parallel([ + () => { c2.abort(new Error('sync stop')); return delay(5) }, + async () => 'b', + ], 2, { signal: c2.signal }), + /sync stop/, + ) + + // Fail-fast abandons unstarted work. + let started = 0 + await assert.rejects(() => parallel(Array.from({ length: 10 }, (_, i) => async () => { + started++ + if (i === 1) { await delay(3); throw new Error('boom') } + await delay(10) + }), 2), /boom/) + const atReject = started + await delay(40) + assert.strictEqual(started, atReject) + + // Infinity = unbounded (2.x behaviour). + let peakInf = 0, activeInf = 0 + await parallel(Array.from({ length: 9 }, () => async () => { + activeInf++; peakInf = Math.max(peakInf, activeInf); await delay(5); activeInf-- + }), Infinity) + assert.strictEqual(peakInf, 9) + + console.log('package-boundary smoke: all checks passed') +})().catch((err) => { console.error(err); process.exit(1) }) diff --git a/test/boundary/smoke.mjs b/test/boundary/smoke.mjs new file mode 100644 index 0000000..0aac2cf --- /dev/null +++ b/test/boundary/smoke.mjs @@ -0,0 +1,24 @@ +// ESM package-boundary smoke test: the default import must be the function, +// not the CJS exports object (the exports-map wrapper guarantees this). +import assert from 'assert' +import parallel, { parallel as named, settle, map, mapSettled, DEFAULT_CONCURRENCY } from 'await-parallel-limit' + +assert.strictEqual(typeof parallel, 'function', 'ESM default import must be the function') +assert.strictEqual(named, parallel) +assert.strictEqual(DEFAULT_CONCURRENCY, 5) + +const results = await parallel([async () => 'esm', async () => 'ok'], 2) +assert.deepStrictEqual(results, ['esm', 'ok']) + +assert.deepStrictEqual(await map([1, 2, 3], (n) => n * 2, 2), [2, 4, 6]) +const s = await settle([async () => 1, async () => { throw new Error('x') }], 2) +assert.deepStrictEqual(s.map((r) => r.status), ['fulfilled', 'rejected']) +const ms = await mapSettled([1], async (n) => n) +assert.strictEqual(ms[0].value, 1) + +const c = new AbortController() +const p = parallel(Array.from({ length: 6 }, () => () => new Promise((r) => setTimeout(r, 20))), 2, { signal: c.signal }) +setTimeout(() => c.abort(new Error('esm abort')), 5) +await assert.rejects(() => p, /esm abort/) + +console.log('ESM package-boundary smoke: all checks passed') diff --git a/test/fuzz.cjs b/test/fuzz.cjs new file mode 100644 index 0000000..ea95953 --- /dev/null +++ b/test/fuzz.cjs @@ -0,0 +1,179 @@ +'use strict' + +/** + * Differential fuzzer. Generates seeded random scenarios (job counts, limits, + * delays, failures, sync throws, aborts at random moments) and checks the + * library against invariants and a per-job reference model: + * + * 1. peak in-flight never exceeds the effective limit + * 2. no-failure, no-abort runs resolve to exactly the reference results, in order + * 3. settle/mapSettled (no abort) resolve to exactly the per-job reference + * outcomes — each job's outcome is timing-independent, so this is exact + * 4. fail-fast rejects with one of the injected errors, and no new jobs + * start after the rejection is delivered + * 5. aborted runs either completed first (resolved) or reject with the abort + * reason, and no new jobs start after the rejection + * + * Run under --unhandled-rejections=strict so any stray rejection is fatal. + * Usage: node --unhandled-rejections=strict test/fuzz.cjs [iterations] [seed] + * On failure it prints the seed + scenario for exact reproduction. + */ + +const assert = require('assert') +const { parallel, settle, map, mapSettled, DEFAULT_CONCURRENCY } = require('../dist/index') + +const ITERATIONS = Number(process.argv[2]) || 300 +const BASE_SEED = Number(process.argv[3]) || 0xC0FFEE + +// xorshift32 — deterministic, seedable. +const prng = (seed) => { + let s = seed >>> 0 || 1 + return () => { + s ^= s << 13; s >>>= 0 + s ^= s >> 17 + s ^= s << 5; s >>>= 0 + return s / 0x100000000 + } +} + +const delay = (ms) => new Promise((r) => setTimeout(r, ms)) + +const effectiveLimit = (limit) => + limit === Infinity ? Infinity : + typeof limit === 'number' && Number.isInteger(limit) && limit > 0 ? limit : DEFAULT_CONCURRENCY + +const makeScenario = (rand) => { + const variants = ['parallel', 'settle', 'map', 'mapSettled'] + const limits = [1, 2, 3, 5, 8, Infinity, 0, -2, 2.5, NaN, undefined] + const n = Math.floor(rand() * 40) + const scenario = { + variant: variants[Math.floor(rand() * variants.length)], + n, + limit: limits[Math.floor(rand() * limits.length)], + jobs: Array.from({ length: n }, (_, i) => ({ + delayMs: Math.floor(rand() * 4), + fails: rand() < 0.08, + syncThrow: rand() < 0.04, + syncAborts: false, + })), + abort: 'none', // none | pre | timed | sync + abortAfterMs: 0, + } + const roll = rand() + if (roll < 0.1) scenario.abort = 'pre' + else if (roll < 0.3) { scenario.abort = 'timed'; scenario.abortAfterMs = Math.floor(rand() * 12) } + else if (roll < 0.38 && n > 0) { scenario.abort = 'sync'; scenario.jobs[Math.floor(rand() * n)].syncAborts = true } + return scenario +} + +const runScenario = async (scenario) => { + const { variant, n, limit, jobs } = scenario + const settleMode = variant === 'settle' || variant === 'mapSettled' + const state = { active: 0, peak: 0, started: 0 } + const controller = scenario.abort === 'none' ? null : new AbortController() + if (scenario.abort === 'pre') controller.abort(new Error('PRE_ABORT')) + + const body = async (spec, i) => { + state.started++ + state.active++ + state.peak = Math.max(state.peak, state.active) + try { + if (spec.delayMs) await delay(spec.delayMs) + if (spec.fails) throw new Error(`FAIL_${i}`) + return i * 2 + 1 + } finally { + state.active-- + } + } + + const thunkFor = (spec, i) => () => { + if (spec.syncAborts && controller) controller.abort(new Error('SYNC_ABORT')) + if (spec.syncThrow) throw new Error(`SYNC_${i}`) + return body(spec, i) + } + + const options = controller ? { signal: controller.signal } : undefined + let promise + if (variant === 'parallel') promise = parallel(jobs.map(thunkFor), limit, options) + else if (variant === 'settle') promise = settle(jobs.map(thunkFor), limit, options) + else { + const mapper = (spec, i) => thunkFor(spec, i)() + promise = variant === 'map' ? map(jobs, mapper, limit, options) : mapSettled(jobs, mapper, limit, options) + } + if (scenario.abort === 'timed') delay(scenario.abortAfterMs).then(() => controller.abort(new Error('TIMED_ABORT'))) + + let outcome + try { + outcome = { resolved: true, value: await promise } + } catch (err) { + outcome = { resolved: false, error: err } + } + const startedAtSettle = state.started + await delay(30) // long enough for any illegally-scheduled job to have started + return { outcome, state, startedAtSettle } +} + +const check = (scenario, { outcome, state, startedAtSettle }) => { + const { n, limit, jobs } = scenario + const settleMode = scenario.variant === 'settle' || scenario.variant === 'mapSettled' + const lim = effectiveLimit(limit) + + // Invariant 1: concurrency ceiling. + assert.ok(state.peak <= Math.min(lim, Math.max(n, 1)), `peak ${state.peak} > limit ${lim}`) + + // Invariant 4/5: nothing starts after the promise settles. + assert.strictEqual(state.started, startedAtSettle, 'jobs started after the promise settled') + + const anyFailure = jobs.some((j) => j.fails || j.syncThrow) + const reference = jobs.map((j, i) => + j.syncThrow || j.fails + ? { status: 'rejected', message: j.syncThrow ? `SYNC_${i}` : `FAIL_${i}` } + : { status: 'fulfilled', value: i * 2 + 1 }) + + if (outcome.resolved) { + // A resolved run must have executed everything: full, ordered, exact. + if (settleMode) { + assert.strictEqual(outcome.value.length, n) + outcome.value.forEach((r, i) => { + assert.strictEqual(r.status, reference[i].status, `settled[${i}] status`) + if (r.status === 'fulfilled') assert.strictEqual(r.value, reference[i].value, `settled[${i}] value`) + else assert.strictEqual(r.reason.message, reference[i].message, `settled[${i}] reason`) + }) + } else { + assert.ok(!anyFailure, 'fail-fast resolved despite an injected failure') + assert.deepStrictEqual(outcome.value, reference.map((r) => r.value)) + } + } else { + const msg = outcome.error && outcome.error.message + const legal = + (scenario.abort !== 'none' && /ABORT/.test(msg)) || + (!settleMode && anyFailure && /^(FAIL|SYNC)_\d+$/.test(msg)) + assert.ok(legal, `illegal rejection: ${msg} (abort=${scenario.abort}, anyFailure=${anyFailure})`) + if (settleMode) { + assert.ok(scenario.abort !== 'none', 'settle mode rejected without an abort') + } + } +} + +;(async () => { + let failures = 0 + for (let iter = 0; iter < ITERATIONS; iter++) { + const seed = (BASE_SEED + iter * 2654435761) >>> 0 + const scenario = makeScenario(prng(seed)) + try { + check(scenario, await runScenario(scenario)) + } catch (err) { + failures++ + console.error(`\n✗ iteration ${iter} seed ${seed}`) + console.error(' scenario:', JSON.stringify(scenario)) + console.error(' ', err.message) + if (failures >= 5) break + } + if ((iter + 1) % 100 === 0) console.log(` ${iter + 1}/${ITERATIONS} scenarios ok`) + } + if (failures) { + console.error(`\nFUZZ FAILED: ${failures} scenario(s). Reproduce: node test/fuzz.cjs 1 `) + process.exit(1) + } + console.log(`\nfuzz: ${ITERATIONS} randomized scenarios passed (base seed ${BASE_SEED.toString(16)})`) +})() diff --git a/test/index.test.cjs b/test/index.test.cjs index 09bbefd..6f8ba02 100644 --- a/test/index.test.cjs +++ b/test/index.test.cjs @@ -2,7 +2,7 @@ const assert = require('assert') const parallel = require('../dist/index').default -const { parallel: named, DEFAULT_CONCURRENCY } = require('../dist/index') +const { parallel: named, settle, map, mapSettled, DEFAULT_CONCURRENCY } = require('../dist/index') const delay = (ms) => new Promise((resolve) => setTimeout(resolve, ms)) @@ -114,6 +114,22 @@ test('treats non-positive / non-integer limits as the default', async () => { } }) +test('limit Infinity means unbounded (2.x behaviour), not the default', async () => { + const { jobs, state } = makeTrackedJobs(12) + await parallel(jobs, Infinity) + assert.strictEqual(state.peak, 12, 'all jobs should run at once') +}) + +test('mutating the input array mid-run does not change the result set', async () => { + const jobs = [ + async () => { jobs.push(async () => 'sneaked-in'); return 1 }, + async () => 2, + async () => 3, + ] + const results = await parallel(jobs, 2) + assert.deepStrictEqual(results, [1, 2, 3], 'appended job must not run or appear') +}) + test('does not start more workers than there are jobs', async () => { const { jobs, state } = makeTrackedJobs(2) await parallel(jobs, 100) @@ -146,6 +162,230 @@ test('handles many jobs without deep recursion', async () => { assert.strictEqual(results[999], 999) }) +// --- v3: options object (3rd positional arg) -------------------------------- + +test('parallel accepts an options object as the 3rd argument', async () => { + const { jobs, state } = makeTrackedJobs(12) + const results = await parallel(jobs, 4, {}) + assert.strictEqual(state.peak, 4) + assert.strictEqual(results.length, 12) +}) + +// --- v3: settle() ----------------------------------------------------------- + +test('settle collects per-job outcomes in order and never rejects', async () => { + const jobs = [ + async () => 'ok', + async () => { throw new Error('bad') }, + async () => { await delay(5); return 42 }, + ] + const results = await settle(jobs, 2) + assert.strictEqual(results[0].status, 'fulfilled') + assert.strictEqual(results[0].value, 'ok') + assert.strictEqual(results[1].status, 'rejected') + assert.strictEqual(results[1].reason.message, 'bad') + assert.strictEqual(results[2].status, 'fulfilled') + assert.strictEqual(results[2].value, 42) +}) + +test('settle respects the concurrency limit', async () => { + const { jobs, state } = makeTrackedJobs(15) + await settle(jobs, 3) + assert.strictEqual(state.peak, 3) +}) + +// --- v3: map() / mapSettled() ---------------------------------------------- + +test('map applies a mapper with concurrency and preserves order', async () => { + let active = 0 + let peak = 0 + const items = [1, 2, 3, 4, 5, 6, 7, 8] + const results = await map(items, async (n, i) => { + active++; peak = Math.max(peak, active) + await delay(5) + active-- + return `${i}:${n * 2}` + }, 3) + assert.strictEqual(peak, 3) + assert.deepStrictEqual(results, ['0:2', '1:4', '2:6', '3:8', '4:10', '5:12', '6:14', '7:16']) +}) + +test('map supports synchronous mappers and fails fast', async () => { + const doubled = await map([1, 2, 3], (n) => n * 2, 2) + assert.deepStrictEqual(doubled, [2, 4, 6]) + await assert.rejects(() => map([1, 2, 3], (n) => { if (n === 2) throw new Error('x'); return n }, 2), /x/) +}) + +test('mapSettled collects per-item outcomes without rejecting', async () => { + const results = await mapSettled([1, 2, 3], async (n) => { + if (n === 2) throw new Error('nope') + return n * 10 + }, 2) + assert.deepStrictEqual(results.map((r) => r.status), ['fulfilled', 'rejected', 'fulfilled']) + assert.strictEqual(results[0].value, 10) + assert.strictEqual(results[1].reason.message, 'nope') + assert.strictEqual(results[2].value, 30) +}) + +test('map and mapSettled reject (not throw synchronously) on non-array input', async () => { + for (const fn of [map, mapSettled]) { + let threwSync = false + let p + try { + p = fn({}, (x) => x, 2) + } catch { + threwSync = true + } + assert.strictEqual(threwSync, false, `${fn.name} must not throw synchronously`) + await assert.rejects(p, /First argument is not an array/) + } +}) + +test('stops starting new jobs after a fail-fast rejection', async () => { + let started = 0 + const jobs = Array.from({ length: 10 }, (_, i) => async () => { + started++ + if (i === 1) { await delay(5); throw new Error('boom') } + await delay(15) + }) + await assert.rejects(() => parallel(jobs, 2), /boom/) + const startedAtRejection = started + await delay(80) // long enough for stragglers to have started more if they were going to + assert.strictEqual(started, startedAtRejection, 'no new jobs may start after rejection') + assert.ok(started < 10, 'the remaining jobs should have been abandoned') +}) + +test('settle resolves to an empty array for empty input', async () => { + assert.deepStrictEqual(await settle([], 3), []) +}) + +test('map and mapSettled treat sparse-array holes as undefined items', async () => { + const results = await map([1, , 3], async (n, i) => (n === undefined ? `hole@${i}` : n * 2), 2) + assert.deepStrictEqual(results, [2, 'hole@1', 6]) + const settled = await mapSettled([1, , 3], async (n) => (n === undefined ? 'hole' : n * 10), 2) + assert.deepStrictEqual(settled.map((r) => r.status), ['fulfilled', 'fulfilled', 'fulfilled']) + assert.strictEqual(settled[1].value, 'hole') +}) + +// --- v3: AbortSignal -------------------------------------------------------- + +/** + * Minimal AbortSignalLike stub: no `reason` support unless abort() is given + * one, and add/remove listener counters so tests can pin cleanup behaviour. + */ +const makeStubSignal = () => { + const listeners = new Set() + const signal = { + aborted: false, + reason: undefined, + adds: 0, + removes: 0, + addEventListener(type, fn) { signal.adds++; listeners.add(fn) }, + removeEventListener(type, fn) { signal.removes++; listeners.delete(fn) }, + abort(reason) { + signal.aborted = true + signal.reason = reason + for (const fn of Array.from(listeners)) fn() + }, + } + return signal +} + +test('rejects immediately when passed an already-aborted signal', async () => { + const controller = new AbortController() + controller.abort(new Error('too late')) + const { jobs } = makeTrackedJobs(5) + await assert.rejects(() => parallel(jobs, 2, { signal: controller.signal }), /too late/) +}) + +test('aborting mid-flight rejects and stops scheduling new jobs', async () => { + const controller = new AbortController() + let started = 0 + const jobs = Array.from({ length: 20 }, () => async () => { + started++ + await delay(20) + }) + const p = parallel(jobs, 3, { signal: controller.signal }) + await delay(25) // let the first wave run + controller.abort(new Error('cancelled')) + await assert.rejects(() => p, /cancelled/) + const startedAtAbort = started + await delay(60) // ensure no further jobs get scheduled after abort + assert.strictEqual(started, startedAtAbort, 'no new jobs should start after abort') + assert.ok(started < 20, 'should not have scheduled all jobs') +}) + +test('native abort() without a reason rejects with the platform AbortError', async () => { + const controller = new AbortController() + const { jobs } = makeTrackedJobs(6, () => delay(15)) + const p = parallel(jobs, 2, { signal: controller.signal }) + await delay(5) + controller.abort() + await assert.rejects(() => p, (err) => err && err.name === 'AbortError') +}) + +test('falls back to a library AbortError for minimal signals without reason', async () => { + const signal = makeStubSignal() + const jobs = Array.from({ length: 6 }, () => () => delay(15)) + const p = parallel(jobs, 2, { signal }) + await delay(5) + signal.abort() // stub keeps reason === undefined → library fallback path + await assert.rejects( + () => p, + (err) => err instanceof Error && err.name === 'AbortError' && /aborted/.test(err.message), + ) +}) + +test('rejects when a job aborts the signal synchronously during startup', async () => { + const controller = new AbortController() + const jobs = [ + () => { controller.abort(new Error('sync abort')); return delay(10).then(() => 'a') }, + async () => 'b', + async () => 'c', + ] + await assert.rejects(() => parallel(jobs, 2, { signal: controller.signal }), /sync abort/) +}) + +test('removes its abort listener after completion and after abort', async () => { + const completed = makeStubSignal() + await parallel([async () => 1, async () => 2], 2, { signal: completed }) + assert.strictEqual(completed.adds, 1) + assert.strictEqual(completed.removes, 1) + + const aborted = makeStubSignal() + const p = parallel(Array.from({ length: 5 }, () => () => delay(15)), 2, { signal: aborted }) + await delay(5) + aborted.abort(new Error('x')) + await assert.rejects(() => p, /x/) + assert.strictEqual(aborted.adds, 1) + assert.strictEqual(aborted.removes, 1) +}) + +test('settle also honours abort (cancellation is not a per-job outcome)', async () => { + const controller = new AbortController() + const jobs = Array.from({ length: 10 }, () => async () => { await delay(20) }) + const p = settle(jobs, 3, { signal: controller.signal }) + await delay(5) + controller.abort(new Error('stop')) + await assert.rejects(() => p, /stop/) +}) + +test('map honours abort', async () => { + const controller = new AbortController() + const p = map(Array.from({ length: 10 }, (_, i) => i), () => delay(20), 3, { signal: controller.signal }) + await delay(5) + controller.abort(new Error('map cancelled')) + await assert.rejects(() => p, /map cancelled/) +}) + +test('a non-aborted signal does not interfere with a normal run', async () => { + const controller = new AbortController() + const { jobs, state } = makeTrackedJobs(8) + const results = await parallel(jobs, 4, { signal: controller.signal }) + assert.strictEqual(state.peak, 4) + assert.strictEqual(results.length, 8) +}) + ;(async () => { for (const [name, fn] of tests) { try { diff --git a/test/types.test.ts b/test/types.test.ts new file mode 100644 index 0000000..01b1b47 --- /dev/null +++ b/test/types.test.ts @@ -0,0 +1,57 @@ +/** + * Type-level regression tests. Compiled (never executed) by `npm test` via + * `tsc --noEmit`; a wrong inference is a build failure. `Equal` distinguishes + * exact types (not mere mutual assignability), so `any` creep fails too. + */ +import parallel, { settle, map, mapSettled, DEFAULT_CONCURRENCY, SettledResult, ParallelOptions } from '../src/index' + +type Equal = (() => T extends A ? 1 : 2) extends (() => T extends B ? 1 : 2) ? true : false +declare function expectType(): void + +async function cases(): Promise { + // Tuple inference from `as const` jobs — the library's signature feature. + const jobs = [ + async () => true, + async () => 2, + async () => 'x', + ] as const + const tuple = await parallel(jobs, 2) + expectType>() + + // Legacy call shapes still typed. + const noLimit = await parallel(jobs) + expectType>() + const withOptions = await parallel(jobs, 2, {} as ParallelOptions) + expectType>() + + // Plain arrays widen to element unions, not any. + const arr: Array<() => Promise> = [async () => 1] + const nums = await parallel(arr, 2) + expectType>() + + // settle: per-slot SettledResult with working discriminated-union narrowing. + const settled = await settle(jobs, 2) + expectType, SettledResult, SettledResult]>>() + const second = settled[1] + if (second.status === 'fulfilled') { + const value = second.value + expectType>() + } else { + const reason = second.reason + expectType>() + } + + // map: mapper drives the result type; sync and async mappers both unwrap. + const mapped = await map([1, 2, 3], (n, i) => `${i}:${n}`, 2) + expectType>() + const mappedAsync = await map([1, 2, 3], async (n) => n * 2) + expectType>() + + // mapSettled. + const ms = await mapSettled([1, 2], async (n) => n > 1) + expectType[]>>() + + expectType>() +} + +void cases