From 9c62f25881f885a977a3fee751b27db2de964886 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 12 Jul 2026 12:25:51 +0000 Subject: [PATCH 1/2] Harden build/publish, algorithm, and tests without changing the API Backward-compatible improvements to the library: - Fix broken publish flow: `prepublish` was outside `scripts` (ignored by npm) and is deprecated. Build now runs via `prepare`, and a `files` allowlist guarantees `dist/` ships even though it is git-ignored. - Correct license metadata (package.json said ISC; LICENSE is MIT). - Rewrite the worker pool as an iterative loop instead of async recursion, avoiding retained promise chains for large job arrays. - Normalise the concurrency limit: non-positive / non-integer values fall back to the default of 5 (preserves the old `limit || 5` behaviour and no longer throws a RangeError on negatives/floats). `limit` is now optional with a default. - Add a named `parallel` export and `DEFAULT_CONCURRENCY` alongside the existing default export (default export unchanged). - Add a real, dependency-free assertion test suite and wire up `npm test` plus a GitHub Actions CI matrix (Node 16/18/20/22). - Modernise tsconfig (drop deprecated/unused options) and refresh README. Tuple return-type inference for `as const` job arrays is preserved. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_013eq96Mbqa2dBCAN3YCyrDK --- .github/workflows/ci.yml | 21 ++++++++ .gitignore | 3 +- README.md | 57 ++++++++++++-------- package-lock.json | 33 ++++++++++++ package.json | 33 +++++++++--- src/index.ts | 71 +++++++++++++++++-------- test/index.test.cjs | 111 +++++++++++++++++++++++++++++++++++++++ test/test.ts | 33 ------------ tsconfig.json | 49 ++++++++--------- 9 files changed, 300 insertions(+), 111 deletions(-) create mode 100644 .github/workflows/ci.yml create mode 100644 package-lock.json create mode 100644 test/index.test.cjs delete mode 100644 test/test.ts diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..655ffa0 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,21 @@ +name: CI + +on: + push: + branches: [master, main] + pull_request: + +jobs: + test: + runs-on: ubuntu-latest + strategy: + matrix: + node-version: [16.x, 18.x, 20.x, 22.x] + steps: + - uses: actions/checkout@v4 + - name: Use Node.js ${{ matrix.node-version }} + uses: actions/setup-node@v4 + with: + node-version: ${{ matrix.node-version }} + - run: npm install + - run: npm test diff --git a/.gitignore b/.gitignore index 86d221c..04c01ba 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,2 @@ node_modules/ -dist/ -test/test.js \ No newline at end of file +dist/ \ No newline at end of file diff --git a/README.md b/README.md index 261bd97..03a414a 100644 --- a/README.md +++ b/README.md @@ -1,48 +1,63 @@ # await-parallel-limit -```javascript +Run an array of async functions with a bounded number running concurrently. +Zero dependencies, first-class TypeScript types. + +```bash npm install await-parallel-limit --save ``` -Processes an array of async functions with concurrency limit (default limit is 5). +## 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`. ## API ```typescript -parallel( - Array jobs, - Integer concurrency, -) +parallel( + jobs: Array<() => Promise>, + concurrency?: number, // default: 5 +): Promise ``` -## Typescript +## TypeScript ```typescript import parallel from 'await-parallel-limit' +// named import also available: import { parallel } from 'await-parallel-limit' const jobs = [ - async () { return await true }, - async () { return await 2 }, + async () => true, + async () => 2, ] as const -// results array is typed based on return types of async functions -// it is an ordered tuple of types - -// const results: [ boolean, number ] +// 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) ``` -## Javascript +## JavaScript ```javascript const parallel = require('await-parallel-limit').default -var jobs = [ - async () { ... }, - async () { ... }, - async () { ... }, - async () { ... } +const jobs = [ + async () => { /* ... */ }, + async () => { /* ... */ }, + async () => { /* ... */ }, + async () => { /* ... */ }, ] -var results = await parallel(jobs, 2) -``` \ No newline at end of file +const results = await parallel(jobs, 2) +``` + +## License + +MIT diff --git a/package-lock.json b/package-lock.json new file mode 100644 index 0000000..83e3fb1 --- /dev/null +++ b/package-lock.json @@ -0,0 +1,33 @@ +{ + "name": "await-parallel-limit", + "version": "2.2.0", + "lockfileVersion": 3, + "requires": true, + "packages": { + "": { + "name": "await-parallel-limit", + "version": "2.2.0", + "license": "MIT", + "devDependencies": { + "typescript": "^5.4.0" + }, + "engines": { + "node": ">=12" + } + }, + "node_modules/typescript": { + "version": "5.9.3", + "resolved": "https://registry.npmjs.org/typescript/-/typescript-5.9.3.tgz", + "integrity": "sha512-jl1vZzPDinLr9eUt3J/t7V6FgNEw9QjvBPdysz9KfQDD41fQrC2Y4vKQdiaUpFT4bXlb1RHhLpp8wtm6M5TgSw==", + "dev": true, + "license": "Apache-2.0", + "bin": { + "tsc": "bin/tsc", + "tsserver": "bin/tsserver" + }, + "engines": { + "node": ">=14.17" + } + } + } +} diff --git a/package.json b/package.json index 0b13ae8..4643452 100644 --- a/package.json +++ b/package.json @@ -1,25 +1,44 @@ { "name": "await-parallel-limit", - "version": "2.1.0", + "version": "2.2.0", "description": "Await an array of async functions with parallel limit.", "main": "dist/index.js", "types": "dist/index.d.ts", + "files": [ + "dist" + ], + "sideEffects": false, "scripts": { - "test": "echo \"Error: no test specified\" && exit 1" + "build": "tsc", + "clean": "rm -rf dist", + "prepare": "tsc", + "test": "tsc && node test/index.test.cjs" }, - "prepublish": "tsc", "keywords": [ "async", "await", "array", "parallel", - "limit" + "concurrency", + "limit", + "promise", + "pool" ], "author": "Ondrej Machek, acrocz@gmail.com", - "license": "ISC", + "license": "MIT", + "engines": { + "node": ">=12" + }, "repository": { "type": "git", - "url": "git://github.com/Acro/await-parallel-limit" + "url": "git+https://github.com/Acro/await-parallel-limit.git" + }, + "bugs": { + "url": "https://github.com/Acro/await-parallel-limit/issues" }, - "dependencies": {} + "homepage": "https://github.com/Acro/await-parallel-limit#readme", + "dependencies": {}, + "devDependencies": { + "typescript": "^5.4.0" + } } diff --git a/src/index.ts b/src/index.ts index 4afb0cb..5db6b9f 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,34 +1,61 @@ type Unpacked = - T extends (...args: any[]) => infer U ? U : - T extends Promise ? U : - T; + T extends (...args: any[]) => infer U ? U : + T extends Promise ? U : + T -const parallel = async ( - jobs: { [K in keyof T]: (() => Promise) }, - limit:number, -) => { - type Data = typeof jobs - type MapToResult = { [K in keyof X]: Unpacked> } +type MapToResult = { [K in keyof T]: Unpacked> } + +export const DEFAULT_CONCURRENCY = 5 +/** + * 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); 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). + */ +const parallel = async ( + jobs: { [K in keyof T]: () => Promise }, + limit: number = DEFAULT_CONCURRENCY, +): Promise> => { if (!Array.isArray(jobs)) { - throw new Error('First argument is not an array.') + throw new Error('First argument is not an array.') } - var n = Math.min(limit || 5, jobs.length) - var ret:any = [] - var index = 0 + // 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 - var next = async () => { - var i = index++ - var job = jobs[i] - ret[i] = await job() - if (index < jobs.length) await next() + const results: any = new Array(jobs.length) + let index = 0 + + const worker = async (): Promise => { + while (true) { + const i = index++ + if (i >= jobs.length) return + results[i] = await jobs[i]() + } } - var next_arr = Array(n).fill(next) - await Promise.all(next_arr.map((f) => { return f() })) + const workerCount = Math.min(concurrency, jobs.length) + const workers: Promise[] = [] + for (let w = 0; w < workerCount; w++) { + workers.push(worker()) + } + await Promise.all(workers) - return >ret + return results as MapToResult } -export default parallel \ No newline at end of file +export default parallel +export { parallel } diff --git a/test/index.test.cjs b/test/index.test.cjs new file mode 100644 index 0000000..23865d7 --- /dev/null +++ b/test/index.test.cjs @@ -0,0 +1,111 @@ +'use strict' + +const assert = require('assert') +const parallel = require('../dist/index').default +const { parallel: named, DEFAULT_CONCURRENCY } = require('../dist/index') + +const delay = (ms) => new Promise((resolve) => setTimeout(resolve, ms)) + +let passed = 0 +const tests = [] +const test = (name, fn) => tests.push([name, fn]) + +/** + * Build a set of jobs that record the peak number of concurrently running jobs, + * so tests can assert the limit is actually respected. + */ +const makeTrackedJobs = (count, work = () => delay(10)) => { + const state = { active: 0, peak: 0 } + const jobs = Array.from({ length: count }, (_, i) => async () => { + state.active++ + state.peak = Math.max(state.peak, state.active) + await work(i) + state.active-- + return i + }) + return { jobs, state } +} + +test('exports a default and a named function plus the default constant', () => { + assert.strictEqual(typeof parallel, 'function') + assert.strictEqual(named, parallel) + assert.strictEqual(DEFAULT_CONCURRENCY, 5) +}) + +test('returns results in input order, not completion order', async () => { + const jobs = [ + async () => { await delay(30); return 'a' }, + async () => { await delay(5); return 'b' }, + async () => { await delay(15); return 'c' }, + ] + const results = await parallel(jobs, 3) + assert.deepStrictEqual(results, ['a', 'b', 'c']) +}) + +test('never exceeds the requested concurrency limit', async () => { + const { jobs, state } = makeTrackedJobs(20) + const results = await parallel(jobs, 4) + assert.strictEqual(state.peak, 4) + assert.deepStrictEqual(results, Array.from({ length: 20 }, (_, i) => i)) +}) + +test('defaults to a concurrency of 5 when the limit is omitted', async () => { + const { jobs, state } = makeTrackedJobs(12) + await parallel(jobs) + assert.strictEqual(state.peak, 5) +}) + +test('treats non-positive / non-integer limits as the default', async () => { + for (const bad of [0, -1, 2.5, NaN]) { + const { jobs, state } = makeTrackedJobs(12) + await parallel(jobs, bad) + assert.strictEqual(state.peak, 5, `limit ${bad} should fall back to 5`) + } +}) + +test('does not start more workers than there are jobs', async () => { + const { jobs, state } = makeTrackedJobs(2) + await parallel(jobs, 100) + assert.strictEqual(state.peak, 2) +}) + +test('resolves to an empty array for empty input', async () => { + const results = await parallel([], 3) + assert.deepStrictEqual(results, []) +}) + +test('propagates the first rejection (Promise.all semantics)', async () => { + const jobs = [ + async () => { await delay(10); return 1 }, + async () => { throw new Error('boom') }, + async () => { await delay(10); return 3 }, + ] + await assert.rejects(() => parallel(jobs, 2), /boom/) +}) + +test('throws synchronously-ish when jobs is not an array', async () => { + await assert.rejects(() => parallel({}, 2), /First argument is not an array/) +}) + +test('handles many jobs without deep recursion', async () => { + const { jobs, state } = makeTrackedJobs(1000, () => delay(0)) + const results = await parallel(jobs, 8) + assert.strictEqual(results.length, 1000) + assert.strictEqual(state.peak, 8) + assert.strictEqual(results[999], 999) +}) + +;(async () => { + for (const [name, fn] of tests) { + try { + await fn() + passed++ + console.log(` ✓ ${name}`) + } catch (err) { + console.error(` ✗ ${name}`) + console.error(err && err.stack ? err.stack : err) + process.exitCode = 1 + } + } + console.log(`\n${passed}/${tests.length} tests passed`) +})() diff --git a/test/test.ts b/test/test.ts deleted file mode 100644 index 3834b1d..0000000 --- a/test/test.ts +++ /dev/null @@ -1,33 +0,0 @@ -import parallel from '../dist/index' - -var count = 0 - -const do_something = async () => { - const timeout = Math.random()*100 + 100 - console.log(count++, timeout) - return await new Promise(resolve => setTimeout(resolve, timeout)) -} - -const another_async = async () => { - return await (async () => { - return [true, true] - })() -} - -const jobs = [ - do_something, - async () => { - return await another_async() - }, - async () => { - return await 2 - }, - do_something, -] as const - -const run = (async () => { - const result = await parallel(jobs, 1) - console.log(result) -}) - -run() \ No newline at end of file diff --git a/tsconfig.json b/tsconfig.json index a351bb7..af49396 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -1,27 +1,24 @@ { - "compilerOptions": { - "declaration": true, - "inlineSourceMap": true, - "inlineSources": true, - "emitDecoratorMetadata": true, - "experimentalDecorators": true, - "module": "commonjs", - "moduleResolution": "node", - "noImplicitAny": true, - "noImplicitReturns": true, - "outDir": "dist", - "removeComments": false, - "rootDir": "src", - "strictNullChecks": true, - "target": "es6", - "baseUrl": "src" - }, - "types": [ "node" ], - "include": [ - "src/**/*" - ], - "exclude": [ - "build" - ] - } - \ No newline at end of file + "compilerOptions": { + "declaration": true, + "inlineSourceMap": true, + "inlineSources": true, + "module": "commonjs", + "moduleResolution": "node", + "strict": true, + "noImplicitAny": true, + "noImplicitReturns": true, + "strictNullChecks": true, + "outDir": "dist", + "removeComments": false, + "rootDir": "src", + "target": "es6" + }, + "include": [ + "src/**/*" + ], + "exclude": [ + "node_modules", + "dist" + ] +} From a805680d470d5d5238fc3da8cf11618eb57befc2 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 12 Jul 2026 13:25:18 +0000 Subject: [PATCH 2/2] test: assert sustained max concurrency (immediate replacement, no batching) Adds a deterministic test using deferred promises: after resolving only the first of two in-flight jobs, the third job must start immediately while the second is still running. This distinguishes a true sliding worker pool from a batching implementation (which would wait for the whole chunk). Verified to fail against a batching impl and pass against the current loop-based pool. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_013eq96Mbqa2dBCAN3YCyrDK --- test/index.test.cjs | 51 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 51 insertions(+) diff --git a/test/index.test.cjs b/test/index.test.cjs index 23865d7..09bbefd 100644 --- a/test/index.test.cjs +++ b/test/index.test.cjs @@ -49,6 +49,57 @@ test('never exceeds the requested concurrency limit', async () => { assert.deepStrictEqual(results, Array.from({ length: 20 }, (_, i) => i)) }) +test('sustains max concurrency: replaces a finished job immediately, no batching', async () => { + // Deferred promises let us control exactly when each job resolves, so this + // test is fully deterministic (no reliance on real-timer timing). + const makeDeferred = () => { + let resolve + const promise = new Promise((r) => { resolve = r }) + return { promise, resolve } + } + // Yield a macrotask so all pending worker continuations (microtasks) run. + const flush = () => new Promise((r) => setTimeout(r, 0)) + + const deferreds = Array.from({ length: 4 }, makeDeferred) + const started = [] + let active = 0 + let peak = 0 + const jobs = deferreds.map((d, i) => async () => { + active++ + peak = Math.max(peak, active) + started.push(i) + await d.promise + active-- + return i + }) + + const done = parallel(jobs, 2) + await flush() + // Only the first two jobs start; the pool is full. + assert.deepStrictEqual(started, [0, 1]) + assert.strictEqual(active, 2) + + // Finish ONLY job 0. Job 1 is still running. A true sliding pool must start + // job 2 right away; a batching impl would wait for job 1 to finish too. + deferreds[0].resolve() + await flush() + assert.deepStrictEqual(started, [0, 1, 2], 'job 2 should start the instant job 0 frees a slot') + assert.strictEqual(active, 2, 'concurrency stays pinned at the limit') + + // Finish job 1 -> job 3 slots in immediately. + deferreds[1].resolve() + await flush() + assert.deepStrictEqual(started, [0, 1, 2, 3]) + assert.strictEqual(active, 2) + + // Drain the tail. + deferreds[2].resolve() + deferreds[3].resolve() + const results = await done + assert.deepStrictEqual(results, [0, 1, 2, 3]) + assert.strictEqual(peak, 2, 'never exceeded the limit') +}) + test('defaults to a concurrency of 5 when the limit is omitted', async () => { const { jobs, state } = makeTrackedJobs(12) await parallel(jobs)