diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml index 7f01ad5..f2bcc32 100644 --- a/.github/workflows/main.yml +++ b/.github/workflows/main.yml @@ -1,36 +1,41 @@ -# This is a basic workflow to help you get started with Actions - name: CI -# Controls when the action will run. Triggers the workflow on push or pull request -# events but only for the main branch on: push: - branches: [ main ] + branches: [main] pull_request: - branches: [ main ] + branches: [main] + +permissions: + contents: read -# A workflow run is made up of one or more jobs that can run sequentially or in parallel jobs: - # This workflow contains a single job called "build" build: - # The type of runner that the job will run on runs-on: ubuntu-latest - - # Steps represent a sequence of tasks that will be executed as part of the job + strategy: + matrix: + node: [22, 24] steps: - # Checks-out your repository under $GITHUB_WORKSPACE, so your job can access it - - uses: actions/checkout@v2 - - name: Use Node.js - uses: actions/setup-node@v1 - with: - node-version: '18.x' - - - name: install deps - run: yarn - - - name: test - run: yarn test - - - name: build lib - run: yarn build \ No newline at end of file + - uses: actions/checkout@v4 + with: + fetch-depth: 0 + - uses: actions/setup-node@v4 + with: + node-version: ${{ matrix.node }} + - name: Install Yarn + run: npm install --global yarn@1.22.22 + - name: Install dependencies + run: yarn install --frozen-lockfile + - name: Typecheck + run: yarn typecheck + - name: Test including performance regression budgets + run: yarn test + - name: Benchmark against the pre-refactor implementation + run: yarn benchmark + - name: Check packed imports and legacy TypeScript consumers + run: node scripts/check-package.cjs + - name: Upload performance measurements + uses: actions/upload-artifact@v4 + with: + name: perf-node-${{ matrix.node }} + path: perf-results.json diff --git a/.gitignore b/.gitignore index cf0fa81..5d63818 100644 --- a/.gitignore +++ b/.gitignore @@ -142,4 +142,6 @@ dist # SvelteKit build / generate output .svelte-kit -# End of https://www.toptal.com/developers/gitignore/api/node \ No newline at end of file +# End of https://www.toptal.com/developers/gitignore/api/node +# Generated performance measurements +perf-results.json diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..866b8f8 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,29 @@ +# Changelog + +## Unreleased + +- Add optional `getTaskKey` for indexed task deduplication and cache lookup. +- Share one task record and result Promise per unique task, and consume serial + queues with a cursor to reduce scheduling work. +- Fix cache expiration and failed-task retries while unrelated tasks are running. +- Settle task failures reliably, including executor and cache-policy exceptions. +- Handle empty dispatches and missing batch results; preserve result order and + duplicate task positions. +- Preserve the configured batch size in serial mode instead of forcing one item. +- Compare cyclic plain objects and arrays safely; use reference identity for + opaque objects such as Map, Set and class instances. +- Correct the built package entry points and include generated TypeScript declarations. +- Retain the published `dist/index.js` and `dist/types.d.ts` paths, CommonJS + `import = require` syntax, and the structural public instance type. Check real + packed consumers in CI. +- Expand behavior and type coverage, add deterministic performance regression + tests, and benchmark against the pre-refactor implementation in CI. +- Rewrite the README around request deduplication, batching and TTL caching, + with runnable examples and clearer cache / execution semantics. +- Update package discovery metadata and include the MIT license text. + +This section describes changes awaiting release. The latest published npm version +is 1.0.1 (2024-03-21). + +The public signatures are retained, but some bug fixes change observable behavior. +See [upgrade notes](docs/upgrading.md) before releasing or upgrading. diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..0c18848 --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2022 Saiya + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/docs/performance.md b/docs/performance.md new file mode 100644 index 0000000..67dc59a --- /dev/null +++ b/docs/performance.md @@ -0,0 +1,67 @@ +# Performance and testing + +[Back to the README](../readme.md) + +This document describes the upcoming release. `getTaskKey` is not available in +npm 1.0.1; see the [changelog](../CHANGELOG.md). +For many tasks or large object inputs, supply a stable, inexpensive `getTaskKey`: + +```ts +import TaskSchedule from 'async-task-schedule' + +const users = new TaskSchedule({ + doTask: async (task: { tenantId: string; userId: string }) => { + const response = await fetch(`/tenants/${encodeURIComponent(task.tenantId)}/users/${encodeURIComponent(task.userId)}`) + if (!response.ok) throw new Error(`HTTP ${response.status}`) + return response.json() + }, + getTaskKey: task => JSON.stringify([task.tenantId, task.userId]), +}) +``` + +Keys take precedence over `isSameTask`. Equal keys share execution and cached results; +the first task's parameters are used. Include every input that affects the result +(e.g. tenant, user, permissions, or query options). Keys must be pure and stable +throughout execution and caching; do not mutate task identity after dispatch. +A symbol must be reused, rather than created anew on each call. Keys use `Map` equality: +`1` differs from `'1'`, `NaN` equals `NaN`, and `0` equals `-0`. +Without `getTaskKey`, deep comparison or your custom `isSameTask` continues to apply. + +Each unique task has one record and a shared result Promise. Duplicate requests +subscribe to that result; completing a task no longer scans every waiting request. +The indexed path avoids linear searches for pending, running, and cached tasks. +Numeric TTLs skip cache scans until the next expiration; function TTL policies are +evaluated on completion and on subsequent cache access. Cache expiration and +failed-task retry eligibility apply even while unrelated work is active. Existing +waiters keep their original result Promise after a cache entry expires. +`cleanCache()` still waits for pending/running work before clearing entries. + +Default deep comparison supports plain objects, arrays (including cyclic values), +Dates, and RegExps. Other objects such as Map, Set, typed arrays, and class instances +use reference identity; supply a task key or custom comparator when those inputs +need semantic equality. An array passed to `dispatch` always means multiple tasks; +wrap an array-valued single task in an object. + +If a TTL policy throws during completion, that task settles with an Error (a single +request rejects, and a multi-task request receives an Error entry). If it throws +on later cache access, the dispatch rejects and that cache entry is removed. +Neither path leaves background rejections or unresolved task records. + +Run `yarn test` for behavior tests, deterministic performance regression budgets, +and coverage; `yarn test:perf` runs just the performance regression suite. +`yarn typecheck` checks source and test/type assertions. `yarn benchmark` builds +the library and compares it with the indexed pre-refactor commit +`f2e07d2c2e358c634a5fb075163fdbc2ef353ccd` in this repository (Git history is required). +It covers batch, individual, overlapping, and serial requests at 100/500/1000 +unique tasks, verifies outputs and execution counts, and records cold/hot-cache +five-run median timings and key-call counts in `perf-results.json`. + +CI on Node 22/24 gates deterministic work budgets: one key extraction per submitted +item, no completion-time identity lookups, and one execution per distinct task. +It also runs the elapsed-time benchmark and uploads its JSON report for comparison. +Elapsed-time measurements include timer and Promise overhead; they are diagnostic, +not machine-independent pass/fail thresholds. No runtime dependencies are added. + +`maxBatchCount` remains the batch size, not a global concurrency limit in parallel +mode. Unlimited caching can retain results indefinitely; use a TTL or `cleanCache()` +when appropriate. The existing batching strategies are preserved. diff --git a/docs/upgrading.md b/docs/upgrading.md new file mode 100644 index 0000000..2bab3bc --- /dev/null +++ b/docs/upgrading.md @@ -0,0 +1,50 @@ +# Upgrading from 1.0.1 + +These notes describe the proposed next release. It has not been published yet. + +## Import and API compatibility + +The constructor options, dispatch overloads, cache-clearing method, running +indicator and static utilities remain available. `getTaskKey` is optional; +existing custom comparators keep working when it is omitted. + +The package retains `dist/index.js`, `dist/index.es.js` and `dist/types.d.ts`. +CommonJS `require('async-task-schedule')`, TypeScript `import Schedule = +require('async-task-schedule')`, and default imports with `esModuleInterop` are +checked against the actual packed package. Exported strategy types and the +structural public instance type remain usable. + +Private runtime fields and the undocumented `defaultOptions` static field are +implementation details. The refactor changes these internals. + +## Observable corrections + +The next release follows the documented behavior more closely. It is not a +guarantee of identical behavior for applications relying on bugs in 1.0.1. + +| Area | 1.0.1 behavior | Next release / upgrade action | +| --- | --- | --- | +| Single-task failure | Could fulfill with an `Error` value. | Rejects. Use `try/catch` or `.catch()`; array dispatch still returns per-item errors. | +| Parallel batch execution | Batches could execute serially even when configured as parallel. | Parallel batches start together. Set `taskExecStrategy: 'serial'` if the backend requires sequential requests. | +| Serial batch size | Serial mode forced a batch size of one. | Honors `maxBatchCount`. Set it to `1` explicitly to preserve one-item batches. | +| Default task comparison | Could treat different Map/Set values as equal; class instances and typed arrays were compared through enumerable fields. | Opaque objects use reference identity. Supply a domain-specific `isSameTask` or stable `getTaskKey` if separate instances should identify the same task. | +| Other equality boundaries | Ignored enumerable symbols, sparse-array lengths and prototype differences; cycles could overflow. | Considers those boundaries and supports cyclic plain objects / arrays. Review custom object inputs or specify identity explicitly. | +| Cache expiration and retries | Could reuse expired or failed entries while unrelated work was active. | Expiration and retry eligibility apply on dispatch even while other work runs. This may cause additional executions that the old bug suppressed. | +| Invalid batch size | Invalid values were not validated reliably. | Nonzero batch sizes must be positive integers; invalid values throw during construction. | +| TTL callback failures | Could escape background work and leave callers unresolved. | Completion failures settle the task as an error; cache-access failures reject dispatch and evict the entry. | + +Cache-policy callbacks should be pure: they can now run at completion as well as +on later cache access. Comparator callbacks should be pure and symmetric; the +indexed implementation does not preserve the old number or order of callback calls. + +The batch-size option does not impose a global concurrency limit in parallel +mode. Keep existing backpressure requirements explicit when upgrading. + +## Validation scope + +Behavior tests cover errors, batching / waiting combinations, deduplication, TTL, +retry and cache clearing. Deterministic performance tests check identity work and +execution counts. Packed-consumer tests check imports, declarations, old file +paths and execution. This verifies supported interfaces and covered behaviors; +it cannot establish compatibility with every application's use of old internals +or previously incorrect behavior. diff --git a/package.json b/package.json index 6a398bf..ec3d745 100644 --- a/package.json +++ b/package.json @@ -1,27 +1,39 @@ { "name": "async-task-schedule", "version": "1.0.1", - "description": "schedule async tasks", - "main": "dist/index.umd.js", + "description": "Batch and deduplicate async requests with TTL caching.", + "main": "dist/index.js", "module": "dist/index.es.js", "types": "dist/types.d.ts", "scripts": { - "build": "vite build", + "build": "vite build && node scripts/compat-package.cjs", + "benchmark": "npm run build && node scripts/benchmark.cjs", + "typecheck": "tsc --noEmit && tsc --project tsconfig.test.json", "test": "vitest run --coverage", - "release": "yarn test && yarn build && yarn publish" + "test:perf": "vitest run test/performance.spec.ts", + "release": "yarn test && yarn build && yarn publish", + "test:package": "npm run build && node scripts/check-package.cjs" }, "keywords": [ - "async task", - "schedule", - "task", + "batch", + "batching", + "deduplication", + "request-deduplication", + "coalescing", + "promise", "async", - "dispatch" + "cache", + "ttl", + "typescript" ], "repository": "https://github.com/oe/async-task-schedule.git", "author": "Saiya ", "license": "MIT", "private": false, "source": "src/index.ts", + "files": [ + "dist" + ], "devDependencies": { "@types/node": "^20.0.0", "@vitest/coverage-v8": "2.1.5", @@ -29,5 +41,9 @@ "vite": "^5.4.11", "vitest": "^2.1.5", "vite-plugin-dts": "^4.3.0" + }, + "homepage": "https://github.com/oe/async-task-schedule#readme", + "bugs": { + "url": "https://github.com/oe/async-task-schedule/issues" } } diff --git a/readme.md b/readme.md index 8aa8380..155c7cf 100644 --- a/readme.md +++ b/readme.md @@ -1,369 +1,260 @@ # async-task-schedule +**Batch and deduplicate async requests with TTL caching.** -
- - github actions - - use it with confident - - code with typescript - - - npm version - - - npm downloads - -
- -schedule async tasks in order - -## Features -* remove duplicated tasks' requests -* combine tasks' requests in same time and do all together -* can prevent massive requests at same time, make them execute one group by one group -* cache result and can specify validity +[![CI](https://github.com/oe/async-task-schedule/actions/workflows/main.yml/badge.svg)](https://github.com/oe/async-task-schedule/actions/workflows/main.yml) +[![npm version](https://img.shields.io/npm/v/async-task-schedule)](https://www.npmjs.com/package/async-task-schedule) +[![MIT license](https://img.shields.io/npm/l/async-task-schedule)](./LICENSE) +Let multiple callers share one execution for the same task. Group distinct tasks +into a batch when your API supports it, and reuse completed results for a configurable +time. Works with synchronous or asynchronous functions, in Node.js and browsers. +TypeScript declarations are included; there are no runtime dependencies. ## Install + ```sh -yarn add async-task-schedule -# or with npm -npm install async-task-schedule -S +npm install async-task-schedule ``` -## Usage +The examples below use existing APIs unless marked **unreleased**. The latest npm +release is currently 1.0.1; `getTaskKey` and the improvements in +[PR #1](https://github.com/oe/async-task-schedule/pull/1) are awaiting publication. +For the default TypeScript imports shown here with npm 1.0.1, enable +`esModuleInterop` in your `tsconfig.json`. -```ts -import TaskSchedule from 'async-task-schedule' -let count = 0 -const taskSchedule = new TaskSchedule({ - doTask: async (name: string) => { - count += 1 - return `${name}${count}` - }, - // or use this will do the same - // batchDoTasks: async (names: string[]) => { - // count += 1 - // return names.map((n) => `${n}${count}`) - // }, -}) +## Quick start: six calls, one batch -taskSchedule.dispatch(['a', 'b']).then(console.log) -taskSchedule.dispatch(['b', 'c']).then(console.log) -taskSchedule.dispatch(['d', 'c']).then(console.log) -taskSchedule.dispatch('c').then(console.log) -// batchDoTasks will be only called once +This example runs without a server. Replace the executor with your batch endpoint +when integrating it into your application. -``` -**NOTICE**: in following example, tasks won't combine ```ts -// batchDoTasks will be executed 3 times due to javascript language features -const result1 = await taskSchedule.dispatch(['a', 'b']) -const result2 = await taskSchedule.dispatch(['b', 'c']) -const result3 = await taskSchedule.dispatch(['d', 'c']) -const result4 = await taskSchedule.dispatch('c') -``` - +import TaskSchedule from 'async-task-schedule' -## API +async function main() { + let requests = 0 + const users = new TaskSchedule({ + batchDoTasks: async (ids: string[]) => { + requests += 1 + console.log('Batch:', ids) + return ids.map(id => ({ id, name: `User ${id}` })) + }, + maxWaitingGap: 10, + invalidAfter: 3000, + }) -### constructor(options: ITaskScheduleOptions) + const results = await Promise.all( + ['a', 'b', 'b', 'c', 'a', 'd'].map(id => users.dispatch(id)), + ) -options define: + console.log(results.map(user => user.id)) // ['a', 'b', 'b', 'c', 'a', 'd'] + console.log(requests) // 1: batch ['a', 'b', 'c', 'd'] -```ts -// `Task` for single task's parameters -// `Result` for single task's response -interface ITaskScheduleOptions { - /** - * action to do batch tasks, can be async or sync function - * Task: single task request info - * Result: single task success response - * - * batchDoTasks should receive multitasks, and return result or error in order - * one of batchDoTasks/doTask must be specified, batchDoTasks will take priority - */ - batchDoTasks: (tasks: Task[]) => Promise> | Array - - /** - * action to do single task, can be async or sync function - * one of batchDoTasks/doTask must be specified, batchDoTasks will take priority - */ - doTask?: (task: Task) => Promise | Result - - /** - * check whether two tasks are equal - * it helps to avoid duplicated tasks - * default: AsyncTask.isEqual (deep equal) - */ - isSameTask?: (a: Task, b: Task) => boolean - - /** - * max task count for batchDoTasks, default unlimited - * undefined or 0 for unlimited - */ - maxBatchCount?: number - - /** - * batch tasks executing strategy, default parallel - * only works if maxBatchCount is specified and tasks more than maxBatchCount are executed - * - * parallel: split all tasks into a list stride by maxBatchCount, exec them at the same time - * serial: split all tasks into a list stride by maxBatchCount, exec theme one group by one group - * if serial specified, when tasks are executing, new comings will wait for them to complete - * it's especially useful to cool down task requests - */ - taskExecStrategy: 'parallel' | 'serial' - - /** - * task waiting strategy, default to debounce - * throttle: tasks will combined and dispatch every `maxWaitingGap` - * debounce: tasks will combined and dispatch util no more tasks in next `maxWaitingGap` - */ - taskWaitingStrategy: 'throttle' | 'debounce' - - /** - * task waiting time in milliseconds, default 50ms - * differently according to taskWaitingStrategy - */ - maxWaitingGap: number - - - /** - * task result caching duration(in milliseconds), default to 1000ms (1s) - * > `undefined` or `0` for unlimited - * > set to minimum value `1` to disable caching - * > `function` to specified specified each task's validity - * - * *cache is lazy cleaned after invalid* - */ - invalidAfter?: number | ((task: Task, result: Result | Error) => number) - - /** - * retry failed tasks next time after failing, default true - */ - retryWhenFailed?: boolean + await users.dispatch('a') // cached; no extra request + console.log(requests) // 1 } -``` - -example: -```ts -import TaskSchedule from 'async-task-schedule' - -const taskSchedule = new TaskSchedule({ - doTask(n) { - console.log(`do task with ${n}`) - return n * n - }, - invalidAfter: 0 -}) - -const result = await taskSchedule.dispatch([1,2,3,1,2]) -// get first result -const resultOf1 = result[0] // 1 -// doTask won't be called -const result11 = await taskSchedule.dispatch(1) // 1 - -// clean all cached result -taskSchedule.cleanCache() -// doTask will be call again -const result12 = await taskSchedule.dispatch(1) // 1 +main().catch(console.error) ``` -### dispatch(tasks: Task[]):Promise> -dispatch multitasks at a time, will get response with corresponding order of `tasks` -this method won't throw any error, it will fulfil even partially failed, you can check whether its success by `response instanceof Error` +Submit calls before awaiting their results to let them join the same waiting window. +Sequential `await` calls cannot join a batch that has already executed, although +those calls can reuse cached results. You can also submit multiple tasks with +`users.dispatch(['a', 'b'])`; results preserve the input order, including duplicates. -```ts -import TaskSchedule from 'async-task-schedule' +## When to use it -const taskSchedule = new TaskSchedule({ - doTask(n) { - console.log(`do task with ${n}`) - if (n % 2) throw new Error(`${n} is unsupported`) - return n * n - }, - invalidAfter: 0 -}) +- Several parts of your application request the same data at nearly the same time. +- Your backend accepts batches, but callers need a convenient single-item function. +- You want a short-lived result cache around an existing async function. +- You want to execute batches serially instead of starting them all together. -const result = await taskSchedule.dispatch([1,2,3,1,2]) -// get first result -const resultOf1 = result[0] // 1 -// second result is error -const isError = result[1] instanceof Error // error object +| Your main need | Consider | +| --- | --- | +| Wrap existing functions with deduplication, batching and built-in TTL | async-task-schedule | +| A mature data loader with batching and request-scoped caching | [DataLoader](https://github.com/graphql/dataloader), which also supports custom scheduling and cache implementations | +| A global concurrency limit or a task queue | [p-limit](https://github.com/sindresorhus/p-limit) or [p-queue](https://github.com/sindresorhus/p-queue) | +| Async function memoization without batching | [p-memoize](https://github.com/sindresorhus/p-memoize) | +| Fetch state, retries and revalidation for a frontend framework | [SWR](https://swr.vercel.app/) or [TanStack Query](https://tanstack.com/query) | +`maxBatchCount` limits the size of each batch. In parallel mode it does **not** impose +a global concurrency limit. The waiting window adds latency (50 ms by default); +choose it according to your application's latency budget. -try { - // will throw an error - const result2 = await taskSchedule.dispatch(2) // 1 -} catch(error) { - console.warn(error) -} -``` +## Recipes +### Deduplicate a JSON read and cache it for three seconds -### dispatch(tasks: Task):Promise -dispatch a task, will get response if success, or throw an error +Use `doTask` when the service has no batch endpoint. Cache parsed data rather than +a raw Fetch `Response`, whose body can only be consumed once. ```ts import TaskSchedule from 'async-task-schedule' -const taskSchedule = new TaskSchedule({ - doTask(n) { - console.log(`do task with ${n}`) - if (n % 2) throw new Error(`${n} is unsupported`) - return n * n +type User = { id: string; name: string } + +// Use a separate instance for each authentication / tenant context. +const users = new TaskSchedule({ + async doTask(id: string): Promise { + const response = await fetch(`/api/users/${encodeURIComponent(id)}`) + if (!response.ok) throw new Error(`HTTP ${response.status}`) + return response.json() }, - invalidAfter: 0 + invalidAfter: 3000, + maxWaitingGap: 0, }) -const result1 = await taskSchedule.dispatch(1) // 1 -try { - const result2 = await taskSchedule.dispatch(2), -} catch(error) { - console.warn(error) -} +export const getUser = (id: string) => users.dispatch(id) ``` -### cleanCache -clean cached tasks' result, so older task will trigger new request and get fresh response. +Use the instance for repeatable reads. A short TTL still shares pending and running +tasks, so it does not make deduplication appropriate for writes that must execute +once per call. Include every result-affecting input in task identity, including +tenant, authorization context and query options. Avoid sharing a cache across users +with different permissions. Treat cached objects as shared data; clone them before +caller-specific mutation. + +### Limit batch size and execute batches serially -attention: this action may not exec immediately, it will take effect after all tasks are done +**Unreleased fix:** the upcoming version preserves `maxBatchCount` in serial mode. +In npm 1.0.1, serial mode forces batches of one item, so this example's batch sizes +require the upcoming release. ```ts import TaskSchedule from 'async-task-schedule' -const taskSchedule = new TaskSchedule({ - doTask(n) { - console.log(`do task with ${n}`) - return n * n - }, - invalidAfter: 0 -}) +async function main() { + const users = new TaskSchedule({ + batchDoTasks: async (ids: string[]) => { + console.log('Batch:', ids) + return ids.map(id => ({ id })) + }, + maxBatchCount: 2, + taskExecStrategy: 'serial', + }) -await Promise.all([ - taskSchedule.dispatch([1, 2, 3, 1, 2]), - taskSchedule.dispatch([1, 9, 10, 12, 22]), -]) -// clean all cached result -taskSchedule.cleanCache() -// task will execute again -const result = await taskSchedule.dispatch(1), + await users.dispatch(['a', 'b', 'c', 'd', 'e']) + // Batches: ['a', 'b'], then ['c', 'd'], then ['e']. +} +main().catch(console.error) ``` -## utils methods -there are some utils method as static members of `async-task-schedule` - +A real batch executor must return one result or `Error` for each input, in the same +order. If your endpoint returns unordered rows, map them back to the requested IDs. +Throwing from the executor fails the entire batch; returning an `Error` entry fails +only that item. Serial execution waits for each batch to finish; it does not enforce +a fixed number of requests per second. -### isEqual(a: unknown, b: unknown): boolean -check whether the given values are equal (with deep comparison) +## API -```ts -import TaskSchedule from 'async-task-schedule' +### `new TaskSchedule(options)` -TaskSchedule.isEqual(1, '1') // false -TaskSchedule.isEqual('1', '1') // true -TaskSchedule.isEqual(NaN, NaN) // true -TaskSchedule.isEqual({a: 'a', b: 'b'}, {b: 'b', a: 'a'}) // true -TaskSchedule.isEqual({a: 'a', b: 'b', c: {e: [1,2,3]}}, {b: 'b', c: {e: [1,2,3]}, a: 'a'}) // true -TaskSchedule.isEqual({a: 'a', b: /acx/}, {b: new RegExp('acx'), a: 'a'}) // true -``` -you can use it to check whether two tasks are equal / find specified task +Supply `doTask`, `batchDoTasks`, or both. If both are supplied, `batchDoTasks` takes priority. +| Option | Default | Behavior | +| --- | --- | --- | +| `doTask(task)` | — | Execute one task; return a value or Promise. | +| `batchDoTasks(tasks)` | — | Execute a batch; return an array or Promise of results / `Error` entries, in input order. | +| `isSameTask(a, b)` | `TaskSchedule.isEqual` | Compare task inputs for deduplication and cache lookup. | +| `getTaskKey(task)` **unreleased** | — | Stable string, number or symbol identity for indexed lookup; takes precedence over `isSameTask`. | +| `maxBatchCount` | Unlimited | `0` or omitted means unlimited; otherwise use a positive integer. | +| `taskExecStrategy` | `'parallel'` | `'parallel'` starts batches together; `'serial'` waits for each batch. In serial mode, `doTask` without a batch size runs one task at a time. | +| `taskWaitingStrategy` | `'debounce'` | `'debounce'` resets the waiting window when new tasks arrive; `'throttle'` keeps a window measured from its first arrival. | +| `maxWaitingGap` | `50` | Waiting-window duration in milliseconds. | +| `invalidAfter` | `1000` | Cache TTL in milliseconds, or `(task, resultOrError) => ttl`. `0` or explicit `undefined` retains results indefinitely. | +| `retryWhenFailed` | `true` | Allow a later dispatch of a failed task to execute again. | -## Receipts +Cache TTL starts when execution completes. Expiration is checked lazily on dispatch; +expired entries are not removed by a background timer. `invalidAfter: 1` means a +one-millisecond TTL, and still allows in-flight deduplication. Use a finite TTL or +`cleanCache()` to avoid retaining an unbounded set of successful results. -### how to integrate with existing code -what you need to do is to wrap your existing task executing function into a new `batchDoTasks` +### `dispatch(task)` and `dispatch(tasks)` -#### example 1: cache `fetch` -suppose we use browser native `fetch` to send request, we can do so to make an improvement: +- A single task resolves to its result, or rejects on failure. +- An array resolves to results or `Error` entries in input order. Task failures do + not reject the array dispatch; identity / cache-policy callback failures can reject it. +- An array always means multiple tasks. Wrap an array-valued single input in an object. ```ts +import TaskSchedule from 'async-task-schedule' -const fetchSchedule = new TaskSchedule({ - async doTask(cfg: {resource: string, options?: RequestInit}) { - return await fetch(cfg.resource, cfg.options) - }, - // 0 for forever - // set a minimum number 1 can disable cache after 1 millisecond - invalidAfter(cfg, result) { - // cache get request for 3s - if (!cfg.options || !cfg.options.method || !cfg.options.method.toLowerCase() === 'get') { - // cache sys static config forever - if (/\/sys\/static-config$/.test(cfg.resource)) return 0 - return 3000 - } - // disable other types request - return 1 - } -}) +async function main() { + const squares = new TaskSchedule({ + doTask(n: number) { + if (n < 0) throw new Error('Expected a non-negative number') + return n * n + }, + maxWaitingGap: 0, + }) -const betterFetch = (resource: string, options?: RequestInit) => { - return fetchSchedule.dispatch({resource, options}) + const results = await squares.dispatch([2, -1, 3]) + console.log(results[0], results[1] instanceof Error, results[2]) // 4, true, 9 + + try { + await squares.dispatch(-1) + } catch (error) { + console.error(error) + } } -// than you can replace fetch with betterFetch +main().catch(console.error) ``` -with those codes above: -1. you can remove redundant request(requests with same parameters at same time will be reduced to one, this may have some side effects) -2. get request can be cached in a short time +### `cleanCache()` +Request clearing of cached results. When work is pending or running, clearing takes +effect after all that work finishes. It does not cancel execution. When idle, +clearing is immediate; the next dispatch executes again. -#### example 2: deal with `getUsers` -suppose we have a method `getUsers` defined as follows: +### Task identity -```ts -// support max 5 users at a time -getUsers(userIds: string[]) => Promise<[{code: string, message: string, id?: string, name?: string, email?: string}]> -``` +Default equality compares plain objects and arrays deeply, including cyclic values, +and compares Dates and RegExps by value. Map, Set, typed arrays and class instances +use reference identity. Supply `isSameTask` for domain-specific equality. These +boundaries describe the upcoming release; 1.0.1 has a less complete comparator. -then we can implement a batch version: -```ts -async function batchGetUsers(userIds: string[]): Promise> { - // there is no need to try/catch, errors will be handled properly - const users = await getUsers(userIds) - // convert invalid users to error - return users.map(user => (user.code === 'failed' ? new Error(user.message) : user)) -} +**Unreleased:** use `getTaskKey` for larger workloads with a natural identity, +for example `task => JSON.stringify([task.tenantId, task.userId])`. Equal keys share +execution and cached results, using the first task's parameters. Include every +input that affects the result. Keys must be pure and stable; avoid mutating tasks +after dispatch. Reuse symbols rather than creating one per call. Keys use `Map` +equality (`1` differs from `'1'`, `NaN` equals `NaN`, `0` equals `-0`). -const getUserSchedule = new TaskSchedule({ - maxBatchCount: 5, - batchDoTasks: batchGetUsers, - // cache user info forever - invalidAfter: 0, -}) +### Static utilities -const result = await Promise.all([ - getUserSchedule.dispatch(['user1', 'user2', 'user3', 'user4', 'user5', 'user6']), - getUserSchedule.dispatch(['user3', 'user2']) - getUserSchedule.dispatch(['user6', 'user7', 'user8', 'user9', 'user10']) - getUserSchedule.dispatch(['user2', 'user6', 'user9']) -]) -// only 2 requests will be sent via getUsers with userIds ['user1', 'user2', 'user3', 'user4', 'user5'] and ['user6', 'user7', 'user8', 'user9', 'user10'] - -// request combine won't works when using await separately -const result1 = await getUserSchedule.dispatch(['user1', 'user2']) -const result2 = await getUserSchedule.dispatch(['user3', 'user2']) -``` +- `TaskSchedule.isEqual(a, b)`: the default equality comparator. +- `TaskSchedule.wrapError(value)`: preserve an `Error`, or wrap another thrown value + in an `Error` with an `original` property. +- `TaskSchedule.runTaskExecutor(executor, ...args)`: resolve to a + `{ status: 'fulfilled', value }` or `{ status: 'rejected', reason }` object. -If you got a batch version function, you just need to make sure it throw an error when error occurred. +## Performance and development +```sh +yarn install --frozen-lockfile +yarn typecheck +yarn test +yarn test:perf +yarn test:package +yarn benchmark +``` -### how to cool down massive requests at the same time -by setting `taskExecStrategy` to `serial` and using smaller `maxBatchCount`(you can even set it to `1`), you can achieve this easily +The performance suite checks deterministic work counts. The benchmark compares the +current implementation with the pre-refactor implementation and writes +`perf-results.json`; elapsed times are diagnostic and depend on the machine. +CI runs on Node 22 and 24 and uploads benchmark reports. -```ts -const taskSchedule = new TaskSchedule({ - ..., - taskExecStrategy: 'serial', - maxBatchCount: 2, -}) -``` +See [performance and testing](https://github.com/oe/async-task-schedule/blob/main/docs/performance.md) +for methodology, task-key guidance and cache-policy behavior. +See the [changelog](https://github.com/oe/async-task-schedule/blob/main/CHANGELOG.md) +for the changes awaiting release. +Before upgrading from 1.0.1, read the +[upgrade notes](https://github.com/oe/async-task-schedule/blob/main/docs/upgrading.md) +for corrected error, batch-execution and equality behavior. Existing import forms +and package paths are retained. +## License +[MIT](./LICENSE) diff --git a/scripts/benchmark.cjs b/scripts/benchmark.cjs new file mode 100644 index 0000000..e8cf05a --- /dev/null +++ b/scripts/benchmark.cjs @@ -0,0 +1,73 @@ +const { performance } = require('node:perf_hooks') +const { execFileSync } = require('node:child_process') +const { writeFileSync } = require('node:fs') +const path = require('node:path') +const assert = require('node:assert/strict') +const ts = require('typescript') +const Current = require('../dist/index.umd.js') + +// This is the tested indexed implementation immediately before the record refactor. +const baselineRef = process.env.PERF_BASELINE_REF || 'f2e07d2c2e358c634a5fb075163fdbc2ef353ccd' +const source = execFileSync('git', ['show', `${baselineRef}:src/index.ts`], { + cwd: path.resolve(__dirname, '..'), encoding: 'utf8', +}) +const compiled = ts.transpileModule(source, { + compilerOptions: { target: ts.ScriptTarget.ES2020, module: ts.ModuleKind.CommonJS }, +}).outputText +const baselineModule = { exports: {} } +// Execute only the baseline source from this repository, using the same Node runtime. +new Function('exports', 'require', 'module', compiled)(baselineModule.exports, require, baselineModule) +const Before = baselineModule.exports.default + +async function sample(Schedule, count, workload) { + let keyCalls = 0 + let executions = 0 + const tasks = Array.from({ length: count }, (_, id) => ({ id, scope: { tenant: 'demo' } })) + const execute = task => { executions++; return task.id } + const schedule = new Schedule({ + ...(workload === 'batch' ? { batchDoTasks: tasks => tasks.map(execute) } : { doTask: execute }), + getTaskKey: task => { keyCalls++; return task.id }, + taskExecStrategy: workload === 'serial' ? 'serial' : 'parallel', + ...(workload === 'serial' ? { maxBatchCount: 1 } : {}), + maxWaitingGap: 0, invalidAfter: 0, retryWhenFailed: false, + }) + const coldStart = performance.now() + if (workload === 'overlap') { + const inputs = tasks.map((task, i) => [task, tasks[(i + 1) % count], task]) + const results = await Promise.all(inputs.map(input => schedule.dispatch(input))) + assert.deepEqual(results, inputs.map(input => input.map(task => task.id))) + } else { + assert.deepEqual(await schedule.dispatch([...tasks, ...tasks]), [...tasks, ...tasks].map(task => task.id)) + } + const coldMs = performance.now() - coldStart + assert.equal(executions, count, 'every distinct task executes exactly once') + const coldKeys = keyCalls + keyCalls = 0 + const hotStart = performance.now() + const cached = await Promise.all(tasks.map(task => schedule.dispatch({ id: task.id, scope: { tenant: 'demo' } }))) + const hotMs = performance.now() - hotStart + assert.deepEqual(cached, tasks.map(task => task.id)) + assert.equal(executions, count, 'hot requests must use cache') + return { coldMs, hotMs, coldKeys, hotKeys: keyCalls } +} + +async function main() { + const rows = [] + for (const workload of ['batch', 'single', 'overlap', 'serial']) { + for (const count of [100, 500, 1000]) { + for (const [version, Schedule] of [['before', Before], ['after', Current]]) { + await sample(Schedule, count, workload) + const samples = [] + for (let i = 0; i < 5; i++) samples.push(await sample(Schedule, count, workload)) + const median = field => samples.map(sample => sample[field]).sort((a, b) => a - b)[2] + rows.push({ workload, tasks: count, version, cold_ms: +median('coldMs').toFixed(3), hot_ms: +median('hotMs').toFixed(3), cold_key_calls: median('coldKeys'), hot_key_calls: median('hotKeys') }) + } + } + } + const report = { node: process.version, baselineRef, warmup: 1, samples: 5, statistic: 'median', rows } + writeFileSync(path.resolve(__dirname, '../perf-results.json'), JSON.stringify(report, null, 2) + '\n') + console.log(`Node ${process.version}; five-run medians after warmup; timings include timer and Promise overhead.`) + console.table(rows) + console.log('Saved perf-results.json. Wall-clock timings are diagnostic; deterministic work budgets gate CI.') +} +main().catch(error => { console.error(error); process.exitCode = 1 }) diff --git a/scripts/check-package.cjs b/scripts/check-package.cjs new file mode 100644 index 0000000..b220dbe --- /dev/null +++ b/scripts/check-package.cjs @@ -0,0 +1,73 @@ +const assert = require('node:assert/strict') +const { execFileSync } = require('node:child_process') +const fs = require('node:fs') +const os = require('node:os') +const path = require('node:path') +const { pathToFileURL } = require('node:url') +const ts = require('typescript') + +const root = path.resolve(__dirname, '..') +const temp = fs.mkdtempSync(path.join(os.tmpdir(), 'async-task-package-')) + +async function main() { + const [packed] = JSON.parse(execFileSync('npm', ['pack', '--json', '--ignore-scripts', '--pack-destination', temp], { + cwd: root, encoding: 'utf8', + })) + const modules = path.join(temp, 'node_modules') + fs.mkdirSync(modules) + execFileSync('tar', ['-xzf', path.join(temp, packed.filename), '-C', modules]) + const pkgDir = path.join(modules, 'async-task-schedule') + fs.renameSync(path.join(modules, 'package'), pkgDir) + const pkg = JSON.parse(fs.readFileSync(path.join(pkgDir, 'package.json'), 'utf8')) + for (const file of [pkg.main, pkg.module, pkg.types, 'dist/index.js', 'dist/types.d.ts', 'LICENSE']) { + assert.ok(fs.existsSync(path.join(pkgDir, file)), `Missing packed file: ${file}`) + } + const Schedule = require(pkgDir) + assert.equal(typeof Schedule, 'function') + assert.equal(typeof require(path.join(pkgDir, 'dist/index.js')), 'function') + const imported = await import(pathToFileURL(path.join(pkgDir, pkg.main)).href) + assert.equal(imported.default, Schedule) + + const sources = [ + `import Schedule = require('async-task-schedule');`, + `import Schedule from 'async-task-schedule';`, + `import Schedule = require('async-task-schedule/dist/types');`, + ] + const fixtures = sources.map((source, i) => { + const file = path.join(temp, `consumer-${i}.ts`) + fs.writeFileSync(file, `${source}\n` + ` + import type { ITaskExecStrategy, ITaskWaitingStrategy, ITaskKey, IAsyncTaskOptions } from 'async-task-schedule'; + const strategy: ITaskExecStrategy = 'serial'; + const waiting: ITaskWaitingStrategy = 'throttle'; + const options: IAsyncTaskOptions = { doTask: n => n * n, taskExecStrategy: strategy, taskWaitingStrategy: waiting }; + const schedule = new Schedule(options); + const scalar: Promise = schedule.dispatch(2); + const tuple: Promise = schedule.dispatch([1, 2] as const); + const instance: Schedule = schedule; + const key: ITaskKey = 'identity'; + // 1.0.1 declarations allowed structural public instances. + const structural: Schedule = { + dispatch: schedule.dispatch, cleanCache() {}, isTaskRunning: false, + }; + `) + return file + }) + const program = ts.createProgram(fixtures, { + target: ts.ScriptTarget.ES2020, module: ts.ModuleKind.CommonJS, + moduleResolution: ts.ModuleResolutionKind.Node10, strict: true, + esModuleInterop: true, skipLibCheck: false, noEmit: true, types: [], + }) + const errors = ts.getPreEmitDiagnostics(program) + assert.equal(errors.length, 0, ts.formatDiagnosticsWithColorAndContext(errors, { + getCanonicalFileName: file => file, getCurrentDirectory: () => temp, getNewLine: () => '\n', + })) + const schedule = new Schedule({ doTask: n => n * n, maxWaitingGap: 0 }) + assert.equal(await schedule.dispatch(3), 9) + assert.deepEqual(await schedule.dispatch([3, 4, 3]), [9, 16, 9]) + console.log('Packed CommonJS / ESM imports, legacy paths, TypeScript consumers and execution pass.') +} + +main().finally(() => fs.rmSync(temp, { recursive: true, force: true })).catch(error => { + console.error(error) + process.exitCode = 1 +}) diff --git a/scripts/compat-package.cjs b/scripts/compat-package.cjs new file mode 100644 index 0000000..f54a7cd --- /dev/null +++ b/scripts/compat-package.cjs @@ -0,0 +1,22 @@ +const { copyFileSync, readFileSync, writeFileSync } = require('node:fs') +const path = require('node:path') + +const dist = path.resolve(__dirname, '../dist') +// Retain the file paths shipped in 1.0.1, including CommonJS deep imports. +copyFileSync(path.join(dist, 'index.umd.js'), path.join(dist, 'index.js')) + +// Preserve the old structural public type and CommonJS `import = require` form. +// The ESM declaration remains available as index.d.ts. +const declaration = readFileSync(path.join(dist, 'index.d.ts'), 'utf8') + .replace(/^ private .+;\r?\n/gm, '') +writeFileSync(path.join(dist, 'index.d.ts'), declaration) +const commonjs = declaration + .replace(/^export default class /m, 'declare class ') + .replace(/^export declare type /gm, 'type ') + .replace(/^export interface /gm, 'interface ') +const names = ['ITaskKey', 'ITaskExecStrategy', 'ITaskWaitingStrategy'] +const namespace = names.map(name => ` type ${name} = import('./index').${name};`).join('\n') +writeFileSync(path.join(dist, 'types.d.ts'), `${commonjs}\n` + + `declare namespace AsyncTask {\n${namespace}\n` + + ` type IAsyncTaskOptions = import('./index').IAsyncTaskOptions;\n` + + `}\nexport = AsyncTask;\n`) diff --git a/src/equal.ts b/src/equal.ts new file mode 100644 index 0000000..7ff2afa --- /dev/null +++ b/src/equal.ts @@ -0,0 +1,28 @@ +/** Structural comparison for plain objects/arrays; opaque objects use identity. */ +export function isEqual(a: unknown, b: unknown, ancestors = new WeakMap()): boolean { + if (a === b) return true + if (typeof a === 'number' && typeof b === 'number' && Number.isNaN(a) && Number.isNaN(b)) return true + if (typeof a !== 'object' || a === null || typeof b !== 'object' || b === null) return false + if (a instanceof RegExp || b instanceof RegExp) { + return a instanceof RegExp && b instanceof RegExp && String(a) === String(b) + } + if (a instanceof Date || b instanceof Date) { + return a instanceof Date && b instanceof Date && isEqual(a.getTime(), b.getTime()) + } + if (Array.isArray(a) !== Array.isArray(b)) return false + if (Array.isArray(a) && Array.isArray(b) && a.length !== b.length) return false + const prototypeA = Object.getPrototypeOf(a) + const prototypeB = Object.getPrototypeOf(b) + if (prototypeA !== prototypeB) return false + if (!Array.isArray(a) && prototypeA !== Object.prototype && prototypeA !== null) return false + const paired = ancestors.get(a) + if (paired) return paired === b + const keysA = Reflect.ownKeys(a).filter(key => Object.prototype.propertyIsEnumerable.call(a, key)) + const keysB = Reflect.ownKeys(b).filter(key => Object.prototype.propertyIsEnumerable.call(b, key)) + if (keysA.length !== keysB.length) return false + ancestors.set(a, b) + const equal = keysA.every(key => Object.prototype.hasOwnProperty.call(b, key) + && isEqual((a as Record)[key], (b as Record)[key], ancestors)) + ancestors.delete(a) + return equal +} diff --git a/src/index.ts b/src/index.ts index 9fccf1c..692453c 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,3 +1,6 @@ +import { isEqual } from './equal' + +export type ITaskKey = string | number | symbol export type ITaskExecStrategy = 'parallel' | 'serial' export type ITaskWaitingStrategy = 'throttle' | 'debounce' export interface IAsyncTaskOptions { @@ -46,371 +49,241 @@ export interface IAsyncTaskOptions { * check whether two tasks are identified the same */ isSameTask?: (a: Task, b: Task) => boolean -} -export default class AsyncTask { - /** - * action to do batch tasks - * Task: single task request info - * Result: single task success response - * - * batchDoTasks should receive multi tasks, and return result or error in order - */ - private batchDoTasks?: (tasks: Task[]) => Promise> | Array - - /** - * do single task - */ - private doTask?: (task: Task) => Result | Error | Promise - - /** - * check whether two tasks are equal - * it helps to avoid duplicated tasks - * default: AsyncTask.isEqual (deep comparison) - */ - private isSameTask: (a: Task, b: Task) => boolean - - /** - * max task count for batchDoTasks, default unlimited - * undefined or 0 for unlimited - */ - private maxBatchCount?: number - - /** - * batch tasks executing strategy, default parallel - * only works if maxBatchCount is specified and tasks more than maxBatchCount are executed - * - * parallel: split all tasks into a list stride by maxBatchCount, exec them at the same time - * serial: split all tasks into a list stride by maxBatchCount, exec theme one group by one group - * if serial specified, when tasks are executing, new comings will wait for them to complete - * it's very useful to cool down task requests - */ - private taskExecStrategy: ITaskExecStrategy - - /** - * task waiting stragy, default to debounce - * throttle: tasks will combined and dispatch every `maxWaitingGap` - * debounce: tasks will combined and dispatch util no more tasks in next `maxWaitingGap` - */ - private taskWaitingStrategy: ITaskWaitingStrategy - - /** - * task waiting time in milliseconds, default 50ms - * differently according to taskWaitingStrategy - */ - private maxWaitingGap: number - - /** - * validity(caching duration) of the result(in ms), default unlimited - * - undefined or 0 for unlimited - * - function to specified each task's validity - * - function receive (task, result) as parameters - * - return a number(ms) as the validity of the result - * - * default to 1s - * - * cache is lazy cleaned after invalid - */ - private invalidAfter?: number | ((task: Task, result: Result | Error) => number) - - /** - * retry failed tasks next time after failing, default true - */ - private retryWhenFailed?: boolean - /** - * tasks ready to be executed + * Optional stable task identity for indexed deduplication and cached lookups. + * Equal keys identify the same task; takes precedence over isSameTask. */ - private pendingTasks: Task[] + getTaskKey?: (task: Task) => ITaskKey +} +interface TaskRecord { + task: Task + key: ITaskKey + state: 'pending' | 'running' | 'done' + promise: Promise + resolve: (value: Result | Error) => void + value?: Result | Error + time: number +} - /** - * original tasks request in queue waiting to resolve - * empty if all task are done - */ - private taskQueue: Array<{ - tasks: Task[]|Task, resolve: Function, reject: Function, isDone?: boolean }> - - /** - * cached task result - */ - private doneTaskMap: Array<{ task: Task, value: Result | Error, time: number }> - - /** - * whether need to clean cache result, aka clean doneTaskMap - */ - private needCleanCache?: boolean - /** - * default task options - */ - private static defaultOptions = { - isSameTask: AsyncTask.isEqual, - taskExecStrategy: 'parallel' as const, - maxWaitingGap: 50, - invalidAfter: 1000, - taskWaitingStrategy: 'debounce' as const, - retryWhenFailed: true, - } +export default class AsyncTask { + private options: IAsyncTaskOptions + private records = new Map>() + private pending: Array> = [] + private pendingHead = 0 + private active = 0 + private pumping = false + private clearRequested = false + private nextExpiry = Infinity + private timeout?: ReturnType + private nextTime?: number + + // Preserve the public running indicator; internal progress uses record counts. + isTaskRunning = false constructor(options: IAsyncTaskOptions) { - const userOptions = { ...AsyncTask.defaultOptions, ...options } - this.pendingTasks = [] - this.doneTaskMap = [] - this.taskQueue = [] - this.isSameTask = userOptions.isSameTask - this.maxBatchCount = userOptions.maxBatchCount - this.maxWaitingGap = userOptions.maxWaitingGap - - this.taskWaitingStrategy = userOptions.taskWaitingStrategy - if (!userOptions.batchDoTasks && !userOptions.doTask) { + if (!options.batchDoTasks && !options.doTask) { throw new Error('one of batchDoTasks / doTask must be specified') } - this.doTask = userOptions.doTask - this.batchDoTasks = userOptions.batchDoTasks - - this.taskExecStrategy = userOptions.taskExecStrategy - if (this.taskExecStrategy === 'serial') { - this.maxBatchCount = 1 + const count = options.maxBatchCount + if (count !== undefined && count !== 0 && (!Number.isInteger(count) || count < 1)) { + throw new Error('maxBatchCount must be a positive integer or 0 for unlimited') + } + this.options = { + ...options, + isSameTask: options.isSameTask ?? AsyncTask.isEqual, + taskExecStrategy: options.taskExecStrategy ?? 'parallel', + taskWaitingStrategy: options.taskWaitingStrategy ?? 'debounce', + maxWaitingGap: options.maxWaitingGap ?? 50, + retryWhenFailed: options.retryWhenFailed ?? true, + // Explicit undefined TTL retains the documented unlimited-cache behavior. + invalidAfter: 'invalidAfter' in options ? options.invalidAfter : 1000, } - - this.retryWhenFailed = userOptions.retryWhenFailed - this.invalidAfter = userOptions.invalidAfter - this.runTasks = this.runTasks.bind(this) this.dispatch = this.dispatch.bind(this) } - /** - * execute task, get task result in promise - */ + async dispatch(task: Task): Promise - /** - * execute tasks, get response in tuple of task and result/error - */ - async dispatch(tasks: T): Promise<{ [k in keyof T]: Result | Error } > - async dispatch(tasks: Task[] | Task) { - this.cleanupTasks() - try { - const result = this.tryGetTaskResult(tasks) - return Promise.resolve(result) - } catch (error) { - // note all tasks are cached, just created new tasks + async dispatch(tasks: T): Promise<{ [k in keyof T]: Result | Error }> + async dispatch(tasks: Task | Task[]): Promise { + this.cleanupExpired() + const requested = Array.isArray(tasks) ? tasks : [tasks] + // Identity callbacks run before changes are committed, and only at dispatch. + const keys = this.options.getTaskKey ? requested.map(task => this.options.getTaskKey!.call(this, task)) : undefined + if (keys && !Array.isArray(tasks)) { + const cached = this.records.get(keys[0]) + if (cached?.state === 'done') return this.unwrap(cached.value!) } - return new Promise((resolve, reject) => { - this.createTasks(tasks, resolve, reject) + const staged = new Map>() + const records = requested.map((task, index) => { + const key = keys ? keys[index] : Symbol() + const existing = keys + ? this.records.get(key) || staged.get(key) + : this.findRecord(task, this.records) || this.findRecord(task, staged) + if (existing) return existing + let resolve!: (value: Result | Error) => void + const promise = new Promise(done => { resolve = done }) + const record: TaskRecord = { task, key, state: 'pending', promise, resolve, time: 0 } + staged.set(key, record) + return record }) + for (const record of staged.values()) { + this.records.set(record.key, record) + this.pending.push(record) + } + if (staged.size) this.schedule() + if (Array.isArray(tasks)) return Promise.all(records.map(record => record.promise)) + const record = records[0] + return this.unwrap(record.state === 'done' ? record.value! : await record.promise) } - /** - * clean cached task result - * - this may not exec immediately - * - it will take effect after all tasks are done - */ - cleanCache() { - this.needCleanCache = true - this.cleanCacheIfNeeded() + private unwrap(value: Result | Error) { + if (value instanceof Error) throw value + return value } - /** - * clean cache if needed - */ - private cleanCacheIfNeeded() { - if (!this.needCleanCache) return - if (this.pendingTasks.length || this.taskQueue.length) return - this.needCleanCache = false - this.doneTaskMap = [] + private findRecord(task: Task, records: Map>) { + for (const record of records.values()) { + if (this.options.isSameTask!.call(this, record.task, task)) return record + } } - /** tasks combine waiting timeout */ - private timeoutId?: any - - /** next exec time for taskWaitingStrategy === 'throttle' */ - private nextTime?: any + /** Cache clearing still waits for all pending and running work. */ + cleanCache() { + this.clearRequested = true + this.clearIfIdle() + } - /** - * create tasks - * @param tasks task list - * @param resolve promise resolve function - * @param reject promise reject function - */ - private createTasks(tasks: Task | Task[], resolve: Function, reject: Function) { - this.taskQueue.push({ tasks, resolve, reject }) - let myTasks = Array.isArray(tasks) ? tasks : [tasks] - // remove duplicated tasks in itself - myTasks = myTasks.filter((task, idx) => idx === myTasks.findIndex(t => this.isSameTask(t, task))) - // remove pending tasks - if (this.pendingTasks.length) { - myTasks = myTasks.filter((f) => !this.hasTask(this.pendingTasks, f)) - } - // remove done tasks - if (myTasks.length) { - myTasks = myTasks.filter((f) => !this.getTaskResult(f)) + private clearIfIdle() { + if (this.clearRequested && !this.active && this.pendingHead === this.pending.length) { + this.records.clear() + this.nextExpiry = Infinity + this.clearRequested = false } - if (!myTasks.length) return - this.pendingTasks = this.pendingTasks.concat(myTasks) + } - clearTimeout(this.timeoutId) - let timeout = 0 - if (this.taskWaitingStrategy === 'throttle') { + private schedule() { + clearTimeout(this.timeout) + let wait = this.options.maxWaitingGap! + if (this.options.taskWaitingStrategy === 'throttle') { const now = Date.now() - this.nextTime = (!this.nextTime || now > this.nextTime) - ? now + this.maxWaitingGap : this.nextTime - timeout = this.nextTime - now - } else { - timeout = this.maxWaitingGap + if (this.nextTime === undefined || now > this.nextTime) this.nextTime = now + wait + wait = this.nextTime - now } - this.timeoutId = setTimeout(this.runTasks, timeout) + this.timeout = setTimeout(() => { void this.run() }, wait) } - // whether task is running - isTaskRunning = false - - private runTasks() { - if (this.isTaskRunning || !this.pendingTasks.length) return - this.isTaskRunning = true - if (this.batchDoTasks) { - this.runTaskWithBatchDoTasks() - } else { - this.runTasksWithDoTask() + /** Consume with a cursor so serial batches do not shift the remaining queue. */ + private take(count: number) { + const records = this.pending.slice(this.pendingHead, this.pendingHead + count) + this.pendingHead += records.length + if (this.pendingHead === this.pending.length) { + this.pending = [] + this.pendingHead = 0 } + for (const record of records) record.state = 'running' + this.active += records.length + this.isTaskRunning = true + return records } - private async runTasksWithDoTask() { - const taskItems = this.pendingTasks.splice(0, this.maxBatchCount || this.pendingTasks.length) - taskItems.forEach(async (task) => { - try { - const result = await this.doTask!(task) - this.updateResultMap([task], [result]) - } catch (error) { - this.updateResultMap([task], AsyncTask.wrapError(error)) - } - this.checkAllTasks() - if (this.pendingTasks.length) { - this.runTasksWithDoTask() - } else { - this.cleanupTasks() - this.isTaskRunning = false - } - }) - } - - private async runTaskWithBatchDoTasks() { - const taskItems = this.pendingTasks.splice(0, this.maxBatchCount || this.pendingTasks.length) + private async run() { + if (this.pumping || this.pendingHead === this.pending.length) return + this.pumping = true try { - const result = await this.batchDoTasks!(taskItems) - this.updateResultMap(taskItems, result) - } catch (error) { - this.updateResultMap(taskItems, AsyncTask.wrapError(error)) - } - this.checkAllTasks() - if (this.pendingTasks.length) { - this.runTaskWithBatchDoTasks() - } else { - this.cleanupTasks() - this.isTaskRunning = false + while (this.pumping && this.pendingHead < this.pending.length) { + const available = this.pending.length - this.pendingHead + const count = this.options.maxBatchCount || (this.options.batchDoTasks ? available : 1) + if (this.options.taskExecStrategy === 'serial') { + await this.execute(this.take(count)) + } else { + const records = this.take(available) + const stride = this.options.maxBatchCount || records.length + const batches: Array> = [] + for (let i = 0; i < records.length; i += stride) batches.push(this.execute(records.slice(i, i + stride))) + await Promise.all(batches) + } + } + } finally { + this.pumping = false + this.isTaskRunning = this.active > 0 + this.clearIfIdle() } } - /** - * check all tasks, try to resolve - */ - private checkAllTasks() { - this.taskQueue.forEach((taskItem) => { + private async execute(records: Array>) { + if (this.options.batchDoTasks) { + let results: Array try { - const result = this.tryGetTaskResult(taskItem.tasks) - // eslint-disable-next-line no-param-reassign - taskItem.isDone = true - taskItem.resolve(result) + results = await this.options.batchDoTasks.call(this, records.map(record => record.task)) + if (!Array.isArray(results)) throw new TypeError('batchDoTasks must return an array') } catch (error) { - // not found + results = records.map(() => AsyncTask.wrapError(error)) } - }) - // clean done task - this.taskQueue = this.taskQueue.filter((task) => !task.isDone) - } - - /** - * get result list of given tasks - * throw error when not found(to make it easier to distinct from falsy results) - * @param tasks tasks to check - * @param defaultResult default result if not found - */ - private tryGetTaskResult(tasks: Task[] | Task): (Result | Error) | Array { - // no cached data and no default result provided - if (!this.doneTaskMap.length) throw new Error('no done task') - - if (Array.isArray(tasks)) { - const result: Array = [] - return tasks.reduce((acc, task) => { - const val = this.getTaskResult(task) - if (!val) throw new Error('not found') - acc.push(val[1]) - return acc - }, result) - } - const val = this.getTaskResult(tasks) - if (!val) throw new Error('not found') - return val[1] - } - - private getTaskResult(task: Task): [Task, Result | Error] | undefined { - const result = this.doneTaskMap.find((t) => this.isSameTask(task, t.task)) - if (result) { - return [result.task, result.value] + let missing: Error | undefined + records.forEach((record, index) => this.complete(record, index < results.length + ? results[index] : (missing || (missing = new Error('not found'))))) + } else { + await Promise.all(records.map(async record => { + let value: Result | Error + try { value = await this.options.doTask!.call(this, record.task) } + catch (error) { value = AsyncTask.wrapError(error) } + this.complete(record, value) + })) } } - private hasTask(list: Task[], task: Task): boolean { - return list.some((item) => this.isSameTask(task, item)) + private validity(record: TaskRecord) { + const ttl = this.options.invalidAfter + return typeof ttl === 'function' ? ttl.call(this, record.task, record.value!) : ttl } - - private updateResultMap(tasks: Task[], result: Array | Error) { - const now = Date.now() - let doneArray: any[] = [] - if (result instanceof Error) { - doneArray = tasks.map((t) => ({ task: t, value: result, time: now })) - } else { - const defaultValue = new Error('not found') - doneArray = tasks.map((t, idx) => { - const taskResult = result.length > idx ? result[idx] : defaultValue - return { task: t, value: taskResult, time: now } - }) + private complete(record: TaskRecord, value: Result | Error) { + record.value = value + record.time = Date.now() + try { + if (value instanceof Error && this.options.retryWhenFailed) { + this.records.delete(record.key) + } else { + const ttl = this.validity(record) + if (ttl && ttl < 0) this.records.delete(record.key) + else if (ttl) this.nextExpiry = Math.min(this.nextExpiry, Math.floor(record.time + ttl) + 1) + } + } catch (error) { + // A policy failure settles this task and cannot escape the background pump. + value = AsyncTask.wrapError(error) + this.records.delete(record.key) + } + record.value = value + record.state = 'done' + --this.active + if (!this.active && this.pendingHead === this.pending.length) { + this.isTaskRunning = false + this.pumping = false } - this.doneTaskMap = this.doneTaskMap.concat(doneArray) + this.clearIfIdle() + record.resolve(value) } - /** - * clean tasks - * - try to clean cache if needed - * - try to remove failed result, remove outdated cache if needed - */ - private cleanupTasks() { - this.cleanCacheIfNeeded() - // has unresolved tasks, unable to cleanup task - if (this.taskQueue.length) return - // nothing to cleanup - if (!this.doneTaskMap.length) return - // no need to remove outdated or failed tasks - if (!this.invalidAfter && !this.retryWhenFailed) return + private cleanupExpired() { const now = Date.now() - this.doneTaskMap = this.doneTaskMap.filter((item) => { - if (this.retryWhenFailed && item.value instanceof Error) { - return false - } - if (this.invalidAfter) { - const invalidAfter = typeof this.invalidAfter === 'function' ? this.invalidAfter(item.task, item.value) : this.invalidAfter - if (!invalidAfter) return true - return now - item.time <= invalidAfter - } - return true - }) + if (typeof this.options.invalidAfter !== 'function' && now < this.nextExpiry) return + this.nextExpiry = Infinity + for (const record of this.records.values()) { + if (record.state !== 'done') continue + let ttl: number | undefined + try { ttl = this.validity(record) } + catch (error) { this.records.delete(record.key); throw error } + if (!ttl) continue + if (now - record.time > ttl) this.records.delete(record.key) + else this.nextExpiry = Math.min(this.nextExpiry, Math.floor(record.time + ttl) + 1) + } } - /** * wrap error info, if it's not instanceof Error, wrap it with Error * @returns Error instance */ static wrapError(e: unknown): Error { if (e instanceof Error) return e - const newError = new Error('task failed') - // @ts-ignore + const newError = new Error('task failed') as Error & { original: unknown } newError.original = e return newError } @@ -434,24 +307,6 @@ export default class AsyncTask { * check whether the given values are equal (with deep comparison) */ static isEqual(a: unknown, b: unknown): boolean { - if (a === b) return true - const typeA = typeof a - const typeB = typeof b - if (typeA !== typeB) return false - // @ts-ignore - // for nan - if (typeA === 'number' && isNaN(a) && isNaN(b)) return true - // none object type, aka primitive types, are checked by the first line - if (typeA !== 'object') return false - // if one of them is regexp, check via regexp literal - if (a instanceof RegExp || b instanceof RegExp) return String(a) === String(b) - if (a instanceof Date || b instanceof Date) return String(a) === String(b) - // only one is array - if (Array.isArray(a) !== Array.isArray(b)) return false - // @ts-ignore - if (Object.keys(a).length !== Object.keys(b).length) return false - // @ts-ignore - if (Object.keys(a).some(k => !AsyncTask.isEqual(a[k], b[k]))) return false - return true + return isEqual(a, b) } } diff --git a/test/async-task-static-methods.spec.ts b/test/async-task-static-methods.spec.ts index 0a7be6c..578e900 100644 --- a/test/async-task-static-methods.spec.ts +++ b/test/async-task-static-methods.spec.ts @@ -113,6 +113,16 @@ describe('async-task static methods', () => { expect(AsyncTask.isEqual(['xx', [1,2,3]], ['xx', [1,2,3]])).toEqual(true) }) + it('handles nulls, own keys, and precise dates', () => { + expect(AsyncTask.isEqual(null, null)).toBe(true) + expect(AsyncTask.isEqual(null, {})).toBe(false) + expect(AsyncTask.isEqual({}, null)).toBe(false) + expect(AsyncTask.isEqual({ a: undefined }, { b: undefined })).toBe(false) + expect(AsyncTask.isEqual(new Date(1), new Date(2))).toBe(false) + expect(AsyncTask.isEqual(new Date(1), {})).toBe(false) + expect(AsyncTask.isEqual(/a/, {})).toBe(false) + }) + it('object', () => { expect(AsyncTask.isEqual({}, {})).toEqual(true) expect(AsyncTask.isEqual({}, /aa/i)).toEqual(false) diff --git a/test/async-task.spec.ts b/test/async-task.spec.ts index f48cff3..56cad6f 100644 --- a/test/async-task.spec.ts +++ b/test/async-task.spec.ts @@ -1,5 +1,7 @@ import AsyncTask from '../src' -import { describe, it, expect } from 'vitest' +import { afterEach, describe, it, expect, vi } from 'vitest' + +afterEach(() => vi.useRealTimers()) function delay(time: number) { return new Promise((resolve) => setTimeout(resolve, time)) @@ -59,12 +61,7 @@ describe('async-task-schedule', () => { ]) // @ts-ignore expect(result[2][0] > 0).toEqual(true) - try { - const result = await at.dispatch(21) - // fail('should go into error') - } catch(e) { - expect(e).toBeInstanceOf(Error) - } + await expect(at.dispatch(21)).rejects.toThrow('jackpot bong! 21') }) }) @@ -129,81 +126,44 @@ describe('async-task-schedule', () => { expect(result[0]).toContain('-result') }) - it('serial waiting', async () => { - const waitTime = 100 - const t1 = Date.now() + it('serial doTask runs one at a time by default', async () => { + let active = 0 + let peak = 0 const at = new AsyncTask({ - // @ts-ignore - batchDoTasks: async (names: number[]) => { - await delay(waitTime) - return names.map((n) => n % 2 ? `${n}-result` : new Error(`not supported ${n}`)).filter(Boolean) + async doTask(n: number) { + peak = Math.max(peak, ++active) + await delay(1) + --active + return n }, - taskExecStrategy: 'serial', - maxBatchCount: 2, - }) - const result = Promise.all([ - at.dispatch([1, 2, 3]), - at.dispatch([3, 4, 5]), - ]) - const result2 = new Promise((resolve) => { - setTimeout(() => { - resolve(at.dispatch([2, 10, 88, 23, 21, 13])) - }, 60); + taskExecStrategy: 'serial', maxWaitingGap: 0, }) - const r = await result - const r1 = await result2 - const t2 = Date.now() - // @ts-ignore - expect(r1.length).toEqual(6) - // make sure all of them exec in serial - expect((t2 - t1) > waitTime * 6).toEqual(true) + await expect(at.dispatch([1, 2, 3])).resolves.toEqual([1, 2, 3]) + expect(peak).toBe(1) }) }) describe('taskWaitingStrategy', () => { - it('throttle waiting', async () => { - const waitTime = 100 - const t1 = Date.now() - const at = new AsyncTask({ - // @ts-ignore - batchDoTasks: async (names: number[]) => { - await delay(waitTime) - return names.map((n) => n % 2 ? `${n}-result` : new Error('not supported')) - }, - taskWaitingStrategy: 'throttle', - maxBatchCount: 2, - }) - const result = Promise.all([ - at.dispatch([1, 2, 3]), - at.dispatch([3, 4, 5]), - ]) - - const delayTime = 60 - const result2 = new Promise((resolve) => { - setTimeout(() => { - resolve(at.dispatch([2, 10, 88, 23, 21, 13])) - }, delayTime); - }) - const r = await result - const r1 = await result2 - const t2 = Date.now() - // @ts-ignore - expect(r1.length).toEqual(6) - const duration = (t2 - t1) - // TODO: not precise - expect( duration > waitTime * 6).toEqual(true) - expect( duration < waitTime * 6 + delayTime + 10).toEqual(true) + it.each(['throttle', 'debounce'] as const)('%s combines requests with the correct waiting window', async (taskWaitingStrategy) => { + vi.useFakeTimers() + const batchDoTasks = vi.fn((tasks: number[]) => tasks) + const at = new AsyncTask({ batchDoTasks, taskWaitingStrategy, maxWaitingGap: 50 }) + const first = at.dispatch(1) + await vi.advanceTimersByTimeAsync(30) + const second = at.dispatch(2) + await vi.advanceTimersByTimeAsync(20) + expect(batchDoTasks).toHaveBeenCalledTimes(taskWaitingStrategy === 'throttle' ? 1 : 0) + await vi.advanceTimersByTimeAsync(30) + await expect(first).resolves.toBe(1) + await expect(second).resolves.toBe(2) + expect(batchDoTasks).toHaveBeenCalledTimes(1) + expect(batchDoTasks).toHaveBeenCalledWith([1, 2]) }) - }) describe('miscs', () => { it('critical parameters missing', () => { - try { - new AsyncTask({}) - } catch (error) { - expect(error).toBeInstanceOf(Error) - } + expect(() => new AsyncTask({})).toThrow('one of batchDoTasks / doTask must be specified') }) it('get cached error result', async () => { @@ -217,27 +177,31 @@ describe('async-task-schedule', () => { retryWhenFailed: false }) const result = await at.dispatch([1,2,3]) - await at.dispatch(1) + await expect(at.dispatch(1)).rejects.toThrow('not implemented') expect(result[2]).toBeInstanceOf(Error) expect(countOf1).toEqual(1) }) - it('get cached error result', async () => { + it('clears cached successes', async () => { + let calls = 0 const at = new AsyncTask({ doTask(n: number) { + ++calls return n * n }, retryWhenFailed: false }) await at.dispatch([1,2,3]) at.cleanCache() - // @ts-ignore - expect(at.doneTaskMap.length).toBe(0) + await expect(at.dispatch(1)).resolves.toBe(1) + expect(calls).toBe(4) }) it('delay clean cache', async () => { + let calls = 0 const at = new AsyncTask({ doTask(n: number) { + ++calls return n * n }, retryWhenFailed: false @@ -247,8 +211,9 @@ describe('async-task-schedule', () => { const result2 = at.dispatch([1,2,3, 7, 0, 1, 10]) await result1 await result2 - // @ts-ignore - expect(at.doneTaskMap.length).toBe(0) + const previousCalls = calls + await expect(at.dispatch([1, 2, 3])).resolves.toEqual([1, 4, 9]) + expect(calls).toBe(previousCalls + 3) }) }) @@ -267,8 +232,6 @@ describe('async-task-schedule', () => { await at.dispatch([1,2,3,4]) await delay(20) await at.dispatch(1) - // @ts-ignore - expect(at.doneTaskMap.length).toEqual(1) expect(task1Count).toEqual(2) }) @@ -285,8 +248,6 @@ describe('async-task-schedule', () => { await at.dispatch([1,2,3,4]) await delay(20) await at.dispatch(1) - // @ts-ignore - expect(at.doneTaskMap.length).toEqual(4) expect(task1Count).toEqual(1) }) @@ -314,8 +275,6 @@ describe('async-task-schedule', () => { await delay(20) await at.dispatch(1) - // @ts-ignore - expect(at.doneTaskMap.length).toEqual(1) expect(callCount[1]).toEqual(1) expect(callCount[2]).toEqual(3) }) diff --git a/test/equality-boundaries.spec.ts b/test/equality-boundaries.spec.ts new file mode 100644 index 0000000..d988277 --- /dev/null +++ b/test/equality-boundaries.spec.ts @@ -0,0 +1,33 @@ +import { expect, it } from 'vitest' +import AsyncTask from '../src' + +it('keeps opaque objects distinct unless they are the same reference', () => { + const map = new Map([['x', 1]]) + expect(AsyncTask.isEqual(map, map)).toBe(true) + expect(AsyncTask.isEqual(map, new Map([['x', 2]]))).toBe(false) + expect(AsyncTask.isEqual(new Set([1]), new Set([2]))).toBe(false) + expect(AsyncTask.isEqual(new Uint8Array([1]), new Uint8Array([1]))).toBe(false) + expect(AsyncTask.isEqual(Promise.resolve(1), Promise.resolve(1))).toBe(false) + expect(AsyncTask.isEqual(Object.create(null), {})).toBe(false) +}) + +it('compares enumerable symbols and sparse array lengths', () => { + const key = Symbol('key') + expect(AsyncTask.isEqual({ [key]: 1 }, { [key]: 2 })).toBe(false) + expect(AsyncTask.isEqual({ [key]: 1 }, { [key]: 1 })).toBe(true) + expect(AsyncTask.isEqual(new Array(1), new Array(2))).toBe(false) + expect(AsyncTask.isEqual(new Array(1), [undefined])).toBe(false) + expect(AsyncTask.isEqual(new Array(1), new Array(1))).toBe(true) + expect(AsyncTask.isEqual(Object.create(null), Object.create(null))).toBe(true) +}) + +it('handles cyclic plain objects without conflating different cycle shapes', () => { + const a: { id: number, self?: unknown } = { id: 1 }; a.self = a + const b: { id: number, self?: unknown } = { id: 1 }; b.self = b + expect(AsyncTask.isEqual(a, b)).toBe(true) + const c = { id: 1, self: b } + expect(AsyncTask.isEqual(a, c)).toBe(false) + const x: unknown[] = []; x.push(x) + const y: unknown[] = []; y.push(y) + expect(AsyncTask.isEqual(x, y)).toBe(true) +}) diff --git a/test/performance.spec.ts b/test/performance.spec.ts new file mode 100644 index 0000000..455bab0 --- /dev/null +++ b/test/performance.spec.ts @@ -0,0 +1,57 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import AsyncTask from '../src' + +beforeEach(() => { vi.useFakeTimers(); vi.setSystemTime(10_000) }) +afterEach(() => vi.useRealTimers()) + +// Work counts are deterministic; elapsed-time benchmarks live in scripts/benchmark.cjs. +describe.each(['single', 'batch'] as const)('performance regression / %s executor', executor => { + it.each([128, 256, 512, 1024])('does one identity lookup per submitted item, N=%s', async count => { + const execute = vi.fn((n: number) => n) + const getTaskKey = vi.fn((n: number) => n) + const isSameTask = vi.fn(() => false) + const schedule = new AsyncTask({ + ...(executor === 'single' ? { doTask: execute } : { batchDoTasks: (tasks: number[]) => tasks.map(execute) }), + getTaskKey, isSameTask, maxWaitingGap: 0, + }) + const tasks = Array.from({ length: count }, (_, i) => i) + const first = schedule.dispatch([...tasks, ...tasks]) + await vi.runAllTimersAsync() + await expect(first).resolves.toEqual([...tasks, ...tasks]) + expect(getTaskKey).toHaveBeenCalledTimes(2 * count) + expect(isSameTask).not.toHaveBeenCalled() + expect(execute).toHaveBeenCalledTimes(count) + getTaskKey.mockClear() + const cached = await Promise.all(tasks.map(task => schedule.dispatch(task))) + expect(cached).toEqual(tasks) + expect(getTaskKey).toHaveBeenCalledTimes(count) + expect(execute).toHaveBeenCalledTimes(count) + }) + + it('shares each completion across 1,000 overlapping requests', async () => { + const execute = vi.fn((n: number) => n) + const getTaskKey = vi.fn((n: number) => n) + const schedule = new AsyncTask({ + ...(executor === 'single' ? { doTask: execute } : { batchDoTasks: (tasks: number[]) => tasks.map(execute) }), + getTaskKey, maxWaitingGap: 0, + }) + const inputs = Array.from({ length: 1000 }, (_, i) => [i % 100, (i + 1) % 100, i % 100]) + const results = inputs.map(tasks => schedule.dispatch(tasks)) + await vi.runAllTimersAsync() + expect(await Promise.all(results)).toEqual(inputs) + expect(execute).toHaveBeenCalledTimes(100) + expect(getTaskKey).toHaveBeenCalledTimes(3000) + }) +}) + +it('has linear identity work for 2,000 serial tasks', async () => { + const getTaskKey = vi.fn((n: number) => n) + const doTask = vi.fn((n: number) => n) + const schedule = new AsyncTask({ getTaskKey, doTask, taskExecStrategy: 'serial', maxBatchCount: 1, maxWaitingGap: 0 }) + const tasks = Array.from({ length: 2000 }, (_, i) => i) + const result = schedule.dispatch(tasks) + await vi.runAllTimersAsync() + await expect(result).resolves.toEqual(tasks) + expect(getTaskKey).toHaveBeenCalledTimes(tasks.length) + expect(doTask).toHaveBeenCalledTimes(tasks.length) +}) diff --git a/test/refactor-regressions.spec.ts b/test/refactor-regressions.spec.ts new file mode 100644 index 0000000..109ea8c --- /dev/null +++ b/test/refactor-regressions.spec.ts @@ -0,0 +1,230 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import AsyncTask from '../src' + +function deferred() { + let resolve!: (value: T) => void + const promise = new Promise(done => { resolve = done }) + return { promise, resolve } +} + +beforeEach(() => { vi.useFakeTimers(); vi.setSystemTime(10_000) }) +afterEach(() => vi.useRealTimers()) + +describe.each([false, true])('shared task records / keyed=%s', keyed => { + const identity = keyed ? { getTaskKey: (n: number) => n } : {} + + it('expires cached results while unrelated work remains active', async () => { + const hold = deferred() + let calls = 0 + const schedule = new AsyncTask({ doTask: (n: number) => n === 2 ? hold.promise : ++calls, invalidAfter: 5, maxWaitingGap: 0, ...identity }) + const first = schedule.dispatch(1) + await vi.advanceTimersByTimeAsync(0) + await expect(first).resolves.toBe(1) + const blocked = schedule.dispatch(2) + await vi.advanceTimersByTimeAsync(6) + const fresh = schedule.dispatch(1) + hold.resolve(2) + await vi.runAllTimersAsync() + await expect(fresh).resolves.toBe(2) + await blocked + expect(calls).toBe(2) + }) + + it('keeps already waiting results even after their cache entries expire', async () => { + const hold = deferred() + const doTask = vi.fn((n: number) => n === 2 ? hold.promise : 10) + const schedule = new AsyncTask({ doTask, invalidAfter: 5, maxWaitingGap: 0, ...identity }) + const original = schedule.dispatch([1, 2, 1]) + await vi.advanceTimersByTimeAsync(6) + const fresh = schedule.dispatch(1) + hold.resolve(20) + await vi.runAllTimersAsync() + await expect(original).resolves.toEqual([10, 20, 10]) + await expect(fresh).resolves.toBe(10) + expect(doTask.mock.calls.map(([n]) => n)).toEqual([1, 2, 1]) + }) + + it('retries a failed task without waiting for unrelated running work', async () => { + const hold = deferred() + let calls = 0 + const error = new Error('temporary') + const schedule = new AsyncTask({ doTask: (n: number) => n === 2 ? hold.promise : ++calls === 1 ? Promise.reject(error) : 10, maxWaitingGap: 0, ...identity }) + const original = schedule.dispatch([1, 2]) + await vi.advanceTimersByTimeAsync(0) + const retry = schedule.dispatch(1) + hold.resolve(20) + await vi.runAllTimersAsync() + await expect(original).resolves.toEqual([error, 20]) + await expect(retry).resolves.toBe(10) + expect(calls).toBe(2) + }) + + it('settles a throwing cache policy as a task error and remains usable', async () => { + const error = new Error('policy failed') + let fail = true + const schedule = new AsyncTask({ doTask: (n: number) => n, invalidAfter: () => { if (fail) throw error; return 0 }, maxWaitingGap: 0, ...identity }) + const first = schedule.dispatch([1, 2]) + await vi.runAllTimersAsync() + await expect(first).resolves.toEqual([error, error]) + expect(schedule.isTaskRunning).toBe(false) + const second = schedule.dispatch(1).catch(reason => reason) + await vi.runAllTimersAsync() + expect(await second).toBe(error) + fail = false + const fresh = schedule.dispatch(1) + await vi.runAllTimersAsync() + await expect(fresh).resolves.toBe(1) + }) + + it('rejects a policy failure on cache access, evicts that entry, then recovers', async () => { + const error = new Error('policy changed') + let fail = false + const schedule = new AsyncTask({ doTask: (n: number) => n, invalidAfter: () => { if (fail) throw error; return 0 }, maxWaitingGap: 0, ...identity }) + const first = schedule.dispatch(1) + await vi.runAllTimersAsync() + await first + fail = true + await expect(schedule.dispatch(1)).rejects.toBe(error) + fail = false + const fresh = schedule.dispatch(1) + await vi.runAllTimersAsync() + await expect(fresh).resolves.toBe(1) + }) + + it('does not retain results with negative TTL', async () => { + const doTask = vi.fn((n: number) => n) + const schedule = new AsyncTask({ doTask, invalidAfter: -1, maxWaitingGap: 0, ...identity }) + for (let i = 0; i < 2; i++) { + const result = schedule.dispatch(1) + await vi.runAllTimersAsync() + await expect(result).resolves.toBe(1) + } + expect(doTask).toHaveBeenCalledTimes(2) + }) +}) + +it('settles invalid batch return values as task errors', async () => { + const schedule = new AsyncTask({ batchDoTasks: (() => null) as unknown as (tasks: number[]) => number[], maxWaitingGap: 0 }) + const result = schedule.dispatch([1, 2]) + await vi.runAllTimersAsync() + const values = await result + expect(values[0]).toBeInstanceOf(TypeError) + expect(values[1]).toBe(values[0]) + expect(schedule.isTaskRunning).toBe(false) +}) + +it('preserves defaults when optional settings are explicitly undefined', async () => { + const schedule = new AsyncTask({ doTask: (n: number) => n, isSameTask: undefined, taskExecStrategy: undefined, taskWaitingStrategy: undefined, maxWaitingGap: undefined, retryWhenFailed: undefined }) + const result = schedule.dispatch([1, 1]) + await vi.advanceTimersByTimeAsync(49) + expect(schedule.isTaskRunning).toBe(false) + await vi.advanceTimersByTimeAsync(1) + await expect(result).resolves.toEqual([1, 1]) +}) + +it('captures keys once and keeps completion independent of key callbacks', async () => { + let executionStarted = false + const getTaskKey = vi.fn((task: { id: number }) => { + if (executionStarted) throw new Error('key extraction during completion') + return task.id + }) + const schedule = new AsyncTask({ getTaskKey, doTask: (task: { id: number }) => { executionStarted = true; return task.id }, maxWaitingGap: 0 }) + const result = schedule.dispatch([{ id: 1 }, { id: 1 }, { id: 2 }]) + await vi.runAllTimersAsync() + await expect(result).resolves.toEqual([1, 1, 2]) + expect(getTaskKey).toHaveBeenCalledTimes(3) +}) + +it('uses a snapshot of the request array', async () => { + const tasks = [1, 2] + const schedule = new AsyncTask({ doTask: (n: number) => n, maxWaitingGap: 0 }) + const result = schedule.dispatch(tasks) + tasks.push(3) + await vi.runAllTimersAsync() + await expect(result).resolves.toEqual([1, 2]) +}) + +it('aborts a comparator failure without committing partial tasks', async () => { + const error = new Error('comparison failed') + const doTask = vi.fn((n: number) => n) + let fail = true + const schedule = new AsyncTask({ doTask, isSameTask: (a, b) => { if (fail) throw error; return a === b }, maxWaitingGap: 0 }) + await expect(schedule.dispatch([1, 2])).rejects.toBe(error) + await vi.runAllTimersAsync() + expect(doTask).not.toHaveBeenCalled() + fail = false + const next = schedule.dispatch([1, 2]) + await vi.runAllTimersAsync() + await expect(next).resolves.toEqual([1, 2]) +}) + +it('handles completion at different times in serial batches without moving the remaining queue', async () => { + const doTask = vi.fn((n: number) => n) + const schedule = new AsyncTask({ doTask, taskExecStrategy: 'serial', maxBatchCount: 1, maxWaitingGap: 0 }) + const result = schedule.dispatch([1, 2, 3, 4]) + await vi.runAllTimersAsync() + await expect(result).resolves.toEqual([1, 2, 3, 4]) + expect(doTask.mock.calls.map(([n]) => n)).toEqual([1, 2, 3, 4]) +}) + +it('retains valid per-task TTL values and respects policy changes on cache access', async () => { + let ttl = 10 + const doTask = vi.fn((n: number) => n) + const schedule = new AsyncTask({ doTask, invalidAfter: () => ttl, maxWaitingGap: 0 }) + const first = schedule.dispatch(1) + await vi.runAllTimersAsync() + await first + await vi.advanceTimersByTimeAsync(3) + await expect(schedule.dispatch(1)).resolves.toBe(1) + ttl = 50 + await vi.advanceTimersByTimeAsync(9) + await expect(schedule.dispatch(1)).resolves.toBe(1) + expect(doTask).toHaveBeenCalledTimes(1) + ttl = 5 + const expired = schedule.dispatch(1) + await vi.runAllTimersAsync() + await expect(expired).resolves.toBe(1) + expect(doTask).toHaveBeenCalledTimes(2) +}) + +it.each(['single', 'batch'] as const)('starts a fresh waiting window for requests chained after %s completion', async executor => { + const execute = vi.fn((n: number) => n) + const schedule = new AsyncTask({ + ...(executor === 'single' ? { doTask: execute } : { batchDoTasks: (tasks: number[]) => tasks.map(execute) }), + maxWaitingGap: 50, getTaskKey: n => n, + }) + const next = schedule.dispatch(1).then(() => schedule.dispatch(2)) + await vi.advanceTimersByTimeAsync(50) + expect(execute.mock.calls.map(([n]) => n)).toEqual([1]) + await vi.advanceTimersByTimeAsync(49) + expect(execute).toHaveBeenCalledTimes(1) + await vi.advanceTimersByTimeAsync(1) + await expect(next).resolves.toBe(2) + expect(execute.mock.calls.map(([n]) => n)).toEqual([1, 2]) +}) + +it.each(['single', 'batch'] as const)('preserves callback receivers in %s mode', async executor => { + let schedule: AsyncTask + const receiver = function (this: AsyncTask, n: number) { + expect(this).toBe(schedule) + return n + } + schedule = new AsyncTask({ + ...(executor === 'single' ? { doTask: receiver } : { batchDoTasks: function (this: AsyncTask, tasks: number[]) { expect(this).toBe(schedule); return tasks } }), + getTaskKey: receiver, + invalidAfter: function (this: AsyncTask) { expect(this).toBe(schedule); return 0 }, + maxWaitingGap: 0, + }) + const result = schedule.dispatch([1, 1]) + await vi.runAllTimersAsync() + await expect(result).resolves.toEqual([1, 1]) + await expect(schedule.dispatch(1)).resolves.toBe(1) + const unkeyed = new AsyncTask({ + doTask: (n: number) => n, + isSameTask: function (this: AsyncTask, a: number, b: number) { expect(this).toBe(unkeyed); return a === b }, + maxWaitingGap: 0, + }) + const duplicate = unkeyed.dispatch([1, 1]) + await vi.runAllTimersAsync() + await expect(duplicate).resolves.toEqual([1, 1]) +}) diff --git a/test/scheduler-behavior.spec.ts b/test/scheduler-behavior.spec.ts new file mode 100644 index 0000000..6b2d4b1 --- /dev/null +++ b/test/scheduler-behavior.spec.ts @@ -0,0 +1,293 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import AsyncTask, { type IAsyncTaskOptions } from '../src' + +function deferred() { + let resolve!: (value: T) => void + let reject!: (reason: unknown) => void + const promise = new Promise((done, fail) => { resolve = done; reject = fail }) + return { promise, resolve, reject } +} + +const modes = (['single', 'batch'] as const).flatMap(executor => + (['serial', 'parallel'] as const).flatMap(strategy => + (['debounce', 'throttle'] as const).flatMap(waiting => + [false, true].map(keyed => ({ executor, strategy, waiting, keyed })), + ), + ), +) + +beforeEach(() => { vi.useFakeTimers(); vi.setSystemTime(10_000) }) +afterEach(() => vi.useRealTimers()) + +describe.each(modes)('$executor / $strategy / $waiting / keyed=$keyed', (mode) => { + function make(execute: (task: number) => number | Promise, extra: Partial> = {}) { + return new AsyncTask({ + ...(mode.executor === 'single' + ? { doTask: execute } + : { batchDoTasks: (tasks) => Promise.all(tasks.map(execute)) }), + taskExecStrategy: mode.strategy, taskWaitingStrategy: mode.waiting, + maxBatchCount: 2, maxWaitingGap: 10, + ...(mode.keyed ? { getTaskKey: (n: number) => n } : {}), + ...extra, + }) + } + + it('preserves request order and repeated positions across overlapping requests', async () => { + const execute = vi.fn((n: number) => n * 10) + const schedule = make(execute) + const first = schedule.dispatch([3, 1, 3, 2]) + const second = schedule.dispatch([2, 4, 1]) + await vi.runAllTimersAsync() + await expect(first).resolves.toEqual([30, 10, 30, 20]) + await expect(second).resolves.toEqual([20, 40, 10]) + expect(execute.mock.calls.map(([n]) => n)).toEqual([3, 1, 2, 4]) + expect(schedule.isTaskRunning).toBe(false) + }) + + it('shares active work and returns ordered values after reverse completion', async () => { + const one = deferred() + const two = deferred() + const execute = vi.fn((n: number) => n === 1 ? one.promise : two.promise) + const schedule = make(execute) + const first = schedule.dispatch([1, 2, 1]) + await vi.advanceTimersByTimeAsync(10) + const second = schedule.dispatch([2, 1]) + const empty = schedule.dispatch([]) + await expect(empty).resolves.toEqual([]) + two.resolve(20) + await vi.advanceTimersByTimeAsync(0) + expect(schedule.isTaskRunning).toBe(true) + one.resolve(10) + await vi.runAllTimersAsync() + await expect(first).resolves.toEqual([10, 20, 10]) + await expect(second).resolves.toEqual([20, 10]) + expect(execute).toHaveBeenCalledTimes(2) + }) + + it('retries failed tasks while keeping successful cached values', async () => { + let attempts = 0 + const error = new Error('temporary failure') + const execute = vi.fn((n: number) => { + if (n === 2 && ++attempts === 1) throw error + return n * 10 + }) + // Individual failures in batch results are represented by Error entries. + const schedule = make(execute, mode.executor === 'batch' ? { + batchDoTasks: (tasks) => Promise.all(tasks.map(async (task) => { + try { return await execute(task) } catch (failure) { return AsyncTask.wrapError(failure) } + })), + } : {}) + const first = schedule.dispatch([1, 2]) + await vi.runAllTimersAsync() + await expect(first).resolves.toEqual([10, error]) + const retry = schedule.dispatch([2, 1, 2]) + await vi.runAllTimersAsync() + await expect(retry).resolves.toEqual([20, 10, 20]) + expect(execute.mock.calls.map(([n]) => n)).toEqual([1, 2, 2]) + }) + + it('keeps TTL boundary values and recomputes them only after expiration', async () => { + const execute = vi.fn((n: number) => n * 10) + const schedule = make(execute, { invalidAfter: 100 }) + const first = schedule.dispatch([1, 2]) + await vi.runAllTimersAsync() + await expect(first).resolves.toEqual([10, 20]) + await vi.advanceTimersByTimeAsync(100) + await expect(schedule.dispatch([2, 1])).resolves.toEqual([20, 10]) + expect(execute).toHaveBeenCalledTimes(2) + await vi.advanceTimersByTimeAsync(1) + const expired = schedule.dispatch([2, 1]) + await vi.runAllTimersAsync() + await expect(expired).resolves.toEqual([20, 10]) + expect(execute).toHaveBeenCalledTimes(4) + }) +}) + +for (const keyed of [false, true]) { + describe(`edge cases / keyed=${keyed}`, () => { + const identity = keyed ? { getTaskKey: (n: number) => n } : {} + + it.each([0, false, '', null, undefined, NaN])('caches a falsy result %s without confusing it with a miss', async (value) => { + const doTask = vi.fn((n: number) => value) + const schedule = new AsyncTask({ doTask, maxWaitingGap: 0, ...identity }) + const first = schedule.dispatch([1, 1]) + await vi.runAllTimersAsync() + await expect(first).resolves.toEqual([value, value]) + await expect(schedule.dispatch(1)).resolves.toEqual(value) + expect(doTask).toHaveBeenCalledTimes(1) + }) + + it.each(['throw', 'reject'] as const)('wraps a non-Error %s and preserves the original value', async (failureMode) => { + const reason = { code: 503 } + const schedule = new AsyncTask({ + doTask: (n: number): Promise => { + if (failureMode === 'throw') throw reason + return Promise.reject(reason) + }, maxWaitingGap: 0, ...identity, + }) + const result = schedule.dispatch(1).then(() => undefined, error => error) + await vi.runAllTimersAsync() + const error = await result + expect(error).toBeInstanceOf(Error) + expect(error.original).toBe(reason) + expect(schedule.isTaskRunning).toBe(false) + }) + + it('keeps cached errors when retry is disabled, then retries after clearing cache', async () => { + const error = new Error('cached failure') + const doTask = vi.fn((n: number): number => { throw error }) + const schedule = new AsyncTask({ doTask, retryWhenFailed: false, invalidAfter: 0, maxWaitingGap: 0, ...identity }) + const first = schedule.dispatch([1]) + await vi.runAllTimersAsync() + await expect(first).resolves.toEqual([error]) + await expect(schedule.dispatch(1)).rejects.toBe(error) + expect(doTask).toHaveBeenCalledTimes(1) + schedule.cleanCache() + const fresh = schedule.dispatch([1]) + await vi.runAllTimersAsync() + await expect(fresh).resolves.toEqual([error]) + expect(doTask).toHaveBeenCalledTimes(2) + }) + + it('supports per-task TTL and unlimited caching in the same result set', async () => { + const doTask = vi.fn((n: number) => n) + const invalidAfter = vi.fn((task: number, result: number | Error) => task === 1 ? 0 : 20) + const schedule = new AsyncTask({ doTask, invalidAfter, maxWaitingGap: 0, ...identity }) + const first = schedule.dispatch([1, 2]) + await vi.runAllTimersAsync() + await first + await vi.advanceTimersByTimeAsync(21) + const next = schedule.dispatch([1, 2]) + await vi.runAllTimersAsync() + await expect(next).resolves.toEqual([1, 2]) + expect(doTask.mock.calls.map(([n]) => n)).toEqual([1, 2, 2]) + expect(invalidAfter).toHaveBeenCalledWith(1, 1) + expect(invalidAfter).toHaveBeenCalledWith(2, 2) + }) + + it('coalesces multiple cache clears while work is active', async () => { + const finish = deferred() + const doTask = vi.fn((n: number) => finish.promise) + const schedule = new AsyncTask({ doTask, invalidAfter: 0, maxWaitingGap: 0, ...identity }) + const first = schedule.dispatch(1) + schedule.cleanCache() + schedule.cleanCache() + await vi.advanceTimersByTimeAsync(0) + const second = schedule.dispatch(1) + finish.resolve(10) + await vi.runAllTimersAsync() + await expect(first).resolves.toBe(10) + await expect(second).resolves.toBe(10) + const next = schedule.dispatch(1) + await vi.runAllTimersAsync() + await expect(next).resolves.toBe(10) + expect(doTask).toHaveBeenCalledTimes(2) + }) + + it.each(['serial', 'parallel'] as const)('isolates a failing batch from successful %s batches', async (taskExecStrategy) => { + const error = new Error('batch failed') + const batchDoTasks = vi.fn((tasks: number[]) => { + if (tasks[0] === 1) throw error + return tasks.map(n => n * 10) + }) + const schedule = new AsyncTask({ batchDoTasks, maxBatchCount: 2, taskExecStrategy, maxWaitingGap: 0, ...identity }) + const result = schedule.dispatch([1, 2, 3, 4, 5]) + await vi.runAllTimersAsync() + await expect(result).resolves.toEqual([error, error, 30, 40, 50]) + expect(batchDoTasks).toHaveBeenCalledTimes(3) + }) + + it('handles missing batch results and ignores surplus results', async () => { + const schedule = new AsyncTask({ batchDoTasks: (tasks: number[]) => tasks[0] === 1 ? [10] : [20, 30, 40], maxWaitingGap: 0, ...identity }) + const missing = schedule.dispatch([1, 2]) + await vi.runAllTimersAsync() + await expect(missing).resolves.toEqual([10, new Error('not found')]) + const extra = schedule.dispatch([2, 3]) + await vi.runAllTimersAsync() + await expect(extra).resolves.toEqual([20, 30]) + }) + + it.each([0, undefined])('supports unlimited batch size %s', async (maxBatchCount) => { + const batchDoTasks = vi.fn((tasks: number[]) => tasks) + const schedule = new AsyncTask({ batchDoTasks, maxBatchCount, maxWaitingGap: 0, ...identity }) + const result = schedule.dispatch([1, 2, 3, 4]) + await vi.runAllTimersAsync() + await expect(result).resolves.toEqual([1, 2, 3, 4]) + expect(batchDoTasks).toHaveBeenCalledTimes(1) + }) + }) +} + +it('prefers the batch executor when both are supplied and keeps dispatch bound', async () => { + const doTask = vi.fn((n: number) => -n) + const batchDoTasks = vi.fn((tasks: number[]) => tasks) + const schedule = new AsyncTask({ doTask, batchDoTasks, maxWaitingGap: 0 }) + const dispatch = schedule.dispatch + const result = dispatch([1, 2]) + await vi.runAllTimersAsync() + await expect(result).resolves.toEqual([1, 2]) + expect(doTask).not.toHaveBeenCalled() +}) + +it.each(['serial', 'parallel'] as const)('lets single-task results settle independently in %s mode', async (taskExecStrategy) => { + const one = deferred() + const two = deferred() + const schedule = new AsyncTask({ doTask: (n: number) => n === 1 ? one.promise : two.promise, taskExecStrategy, maxBatchCount: 2, maxWaitingGap: 0 }) + const first = schedule.dispatch(1) + const second = schedule.dispatch(2) + const combined = schedule.dispatch([1, 2]) + let combinedDone = false + combined.then(() => { combinedDone = true }) + await vi.advanceTimersByTimeAsync(0) + one.resolve(10) + await vi.advanceTimersByTimeAsync(0) + await expect(first).resolves.toBe(10) + expect(combinedDone).toBe(false) + expect(schedule.isTaskRunning).toBe(true) + two.resolve(20) + await vi.runAllTimersAsync() + await expect(second).resolves.toBe(20) + await expect(combined).resolves.toEqual([10, 20]) +}) + +it.each(['serial', 'parallel'] as const)('respects active batch concurrency in %s mode', async (taskExecStrategy) => { + const finish = deferred() + let active = 0 + let peak = 0 + const batchDoTasks = vi.fn(async (tasks: number[]) => { + peak = Math.max(peak, ++active) + await finish.promise + --active + return tasks + }) + const schedule = new AsyncTask({ batchDoTasks, taskExecStrategy, maxBatchCount: 2, maxWaitingGap: 0, getTaskKey: n => n }) + const result = schedule.dispatch([1, 2, 3, 4, 5]) + await vi.advanceTimersByTimeAsync(0) + expect(active).toBe(taskExecStrategy === 'serial' ? 1 : 3) + finish.resolve() + await vi.runAllTimersAsync() + await expect(result).resolves.toEqual([1, 2, 3, 4, 5]) + expect(peak).toBe(taskExecStrategy === 'serial' ? 1 : 3) + expect(batchDoTasks.mock.calls.map(([tasks]) => tasks)).toEqual([[1, 2], [3, 4], [5]]) +}) + +it.each(['debounce', 'throttle'] as const)('%s handles sustained arrivals and a subsequent waiting window', async (taskWaitingStrategy) => { + const batchDoTasks = vi.fn((tasks: number[]) => tasks) + const schedule = new AsyncTask({ batchDoTasks, taskWaitingStrategy, maxWaitingGap: 30, getTaskKey: n => n }) + const responses = [schedule.dispatch(0)] + for (let i = 1; i <= 2; i++) { + await vi.advanceTimersByTimeAsync(10) + responses.push(schedule.dispatch(i)) + } + await vi.advanceTimersByTimeAsync(10) + expect(batchDoTasks).toHaveBeenCalledTimes(taskWaitingStrategy === 'throttle' ? 1 : 0) + await vi.advanceTimersByTimeAsync(20) + expect(await Promise.all(responses)).toEqual([0, 1, 2]) + expect(batchDoTasks).toHaveBeenCalledTimes(1) + const next = schedule.dispatch(3) + await vi.advanceTimersByTimeAsync(29) + expect(batchDoTasks).toHaveBeenCalledTimes(1) + await vi.advanceTimersByTimeAsync(1) + await expect(next).resolves.toBe(3) + expect(batchDoTasks).toHaveBeenCalledTimes(2) +}) diff --git a/test/scheduler-regressions.spec.ts b/test/scheduler-regressions.spec.ts new file mode 100644 index 0000000..a653db0 --- /dev/null +++ b/test/scheduler-regressions.spec.ts @@ -0,0 +1,122 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import AsyncTask from '../src' + +function deferred() { + let resolve!: (value: T) => void + const promise = new Promise((done) => { resolve = done }) + return { promise, resolve } +} + +afterEach(() => vi.useRealTimers()) + +describe('scheduler regressions', () => { + it('resolves empty requests without starting an executor', async () => { + vi.useFakeTimers() + const doTask = vi.fn((n: number) => n) + const schedule = new AsyncTask({ doTask }) + let result: unknown + schedule.dispatch([]).then((value) => { result = value }) + await vi.runAllTimersAsync() + expect(result).toEqual([]) + expect(doTask).not.toHaveBeenCalled() + }) + + it('rejects single failures, including cached failures, but resolves batch errors', async () => { + const error = new Error('failed') + const doTask = vi.fn((n: number): number => { throw error }) + const schedule = new AsyncTask({ doTask, retryWhenFailed: false, invalidAfter: 0, maxWaitingGap: 0 }) + await expect(schedule.dispatch(1)).rejects.toBe(error) + await expect(schedule.dispatch(1)).rejects.toBe(error) + await expect(schedule.dispatch([1])).resolves.toEqual([error]) + expect(doTask).toHaveBeenCalledTimes(1) + }) + + it.each(['single', 'batch'] as const)('deduplicates running %s tasks and stays running until all finish', async (mode) => { + vi.useFakeTimers() + const first = deferred() + const second = deferred() + const executor = vi.fn((n: number) => n === 1 ? first.promise : second.promise) + const schedule = new AsyncTask({ + ...(mode === 'single' ? { doTask: executor } : { batchDoTasks: (tasks: number[]) => Promise.all(tasks.map(executor)) }), + maxBatchCount: 1, maxWaitingGap: 0, + }) + const original = schedule.dispatch([1, 2]) + await vi.advanceTimersByTimeAsync(0) + const duplicate = schedule.dispatch(1) + first.resolve(10) + await vi.advanceTimersByTimeAsync(0) + expect(schedule.isTaskRunning).toBe(true) + second.resolve(20) + await expect(original).resolves.toEqual([10, 20]) + await expect(duplicate).resolves.toBe(10) + expect(executor.mock.calls.map(([n]) => n)).toEqual([1, 2]) + expect(schedule.isTaskRunning).toBe(false) + }) + + it('starts parallel batches together', async () => { + vi.useFakeTimers() + const finish = deferred() + const batchDoTasks = vi.fn(async (tasks: number[]) => { await finish.promise; return tasks }) + const schedule = new AsyncTask({ batchDoTasks, maxBatchCount: 2, maxWaitingGap: 0 }) + const result = schedule.dispatch([1, 2, 3, 4]) + await vi.advanceTimersByTimeAsync(0) + expect(batchDoTasks.mock.calls.map(([tasks]) => tasks)).toEqual([[1, 2], [3, 4]]) + finish.resolve() + await expect(result).resolves.toEqual([1, 2, 3, 4]) + }) + + it('preserves serial batch sizes and queues new requests behind active work', async () => { + vi.useFakeTimers() + const finish = deferred() + const batchDoTasks = vi.fn(async (tasks: number[]) => { await finish.promise; return tasks }) + const schedule = new AsyncTask({ batchDoTasks, taskExecStrategy: 'serial', maxBatchCount: 2, maxWaitingGap: 0 }) + const original = schedule.dispatch([1, 2, 3]) + await vi.advanceTimersByTimeAsync(0) + const later = schedule.dispatch([2, 4]) + await vi.advanceTimersByTimeAsync(0) + expect(batchDoTasks.mock.calls.map(([tasks]) => tasks)).toEqual([[1, 2]]) + finish.resolve() + await expect(original).resolves.toEqual([1, 2, 3]) + await expect(later).resolves.toEqual([2, 4]) + expect(batchDoTasks.mock.calls.map(([tasks]) => tasks)).toEqual([[1, 2], [3, 4]]) + }) + + it('uses an unlimited serial batch when no batch size is supplied', async () => { + const batchDoTasks = vi.fn((tasks: number[]) => tasks) + const schedule = new AsyncTask({ batchDoTasks, taskExecStrategy: 'serial', maxWaitingGap: 0 }) + await expect(schedule.dispatch([1, 2, 3])).resolves.toEqual([1, 2, 3]) + expect(batchDoTasks).toHaveBeenCalledTimes(1) + expect(batchDoTasks).toHaveBeenCalledWith([1, 2, 3]) + }) + + it('defers cache cleaning until every active request finishes', async () => { + vi.useFakeTimers() + const finish = deferred() + const doTask = vi.fn((n: number) => n === 1 ? 10 : finish.promise) + const schedule = new AsyncTask({ doTask, invalidAfter: 0, maxWaitingGap: 0 }) + const first = schedule.dispatch(1) + const second = schedule.dispatch(2) + await vi.advanceTimersByTimeAsync(0) + await expect(first).resolves.toBe(10) + schedule.cleanCache() + const duplicate = schedule.dispatch(1) + await expect(duplicate).resolves.toBe(10) + finish.resolve(20) + await expect(second).resolves.toBe(20) + const fresh = schedule.dispatch(1) + await vi.advanceTimersByTimeAsync(0) + await expect(fresh).resolves.toBe(10) + expect(doTask.mock.calls.map(([n]) => n)).toEqual([1, 2, 1]) + }) + + it('dispatches distinct null and object tasks without comparator errors', async () => { + const doTask = vi.fn((task: null | { id: number }) => task === null ? 0 : task.id) + const schedule = new AsyncTask({ doTask, maxWaitingGap: 0 }) + await expect(schedule.dispatch([null, { id: 1 }, null])).resolves.toEqual([0, 1, 0]) + expect(doTask).toHaveBeenCalledTimes(2) + }) + + it.each([-1, 0.5, NaN, Infinity])('rejects invalid maxBatchCount %s', (maxBatchCount) => { + expect(() => new AsyncTask({ doTask: (n: number) => n, maxBatchCount })).toThrow(/maxBatchCount/) + }) +}) diff --git a/test/scheduler-types.spec.ts b/test/scheduler-types.spec.ts new file mode 100644 index 0000000..8930f04 --- /dev/null +++ b/test/scheduler-types.spec.ts @@ -0,0 +1,29 @@ +import { expectTypeOf, it } from 'vitest' +import AsyncTask, { type IAsyncTaskOptions, type ITaskKey } from '../src' + +it('preserves single, array, readonly tuple, and empty tuple response types', () => { + const schedule = new AsyncTask({ doTask: (n: number) => String(n), getTaskKey: n => n }) + const single = () => schedule.dispatch(1) + const array = () => schedule.dispatch([1, 2] as number[]) + const tuple = () => schedule.dispatch([1, 2] as const) + const empty = () => schedule.dispatch([]) + expectTypeOf>().toEqualTypeOf>() + expectTypeOf>().toEqualTypeOf>>() + expectTypeOf>().toEqualTypeOf>() + expectTypeOf>().toEqualTypeOf>() + expectTypeOf().toEqualTypeOf() +}) + +it('accepts typed key extractors and rejects unsupported key types', () => { + const options: IAsyncTaskOptions<{ id: string }, number> = { + doTask: task => task.id.length, + getTaskKey: task => task.id, + } + expectTypeOf(options.getTaskKey).toEqualTypeOf<((task: { id: string }) => ITaskKey) | undefined>() + const invalid: IAsyncTaskOptions = { + doTask: n => n, + // @ts-expect-error Keys must be stable primitive values, not objects. + getTaskKey: n => ({ id: n }), + } + expectTypeOf(invalid).toMatchTypeOf>() +}) diff --git a/test/task-identity.spec.ts b/test/task-identity.spec.ts new file mode 100644 index 0000000..563f0e8 --- /dev/null +++ b/test/task-identity.spec.ts @@ -0,0 +1,114 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import AsyncTask, { type ITaskKey } from '../src' + +beforeEach(() => { vi.useFakeTimers(); vi.setSystemTime(10_000) }) +afterEach(() => vi.useRealTimers()) + +describe('task identity and performance invariants', () => { + it('uses stable keys instead of object identity or deep comparison', async () => { + const isSameTask = vi.fn(() => false) + const doTask = vi.fn((task: { id: number, label: string }) => task.label) + const schedule = new AsyncTask({ doTask, getTaskKey: task => task.id, isSameTask, maxWaitingGap: 0 }) + const first = schedule.dispatch([{ id: 1, label: 'first' }, { id: 1, label: 'second' }, { id: 2, label: 'other' }]) + await vi.runAllTimersAsync() + await expect(first).resolves.toEqual(['first', 'first', 'other']) + await expect(schedule.dispatch({ id: 1, label: 'cached' })).resolves.toBe('first') + expect(doTask).toHaveBeenCalledTimes(2) + expect(isSameTask).not.toHaveBeenCalled() + }) + + it('keeps numeric, string, symbol, zero, and empty-string keys distinct', async () => { + const a = Symbol('same label') + const b = Symbol('same label') + const keys: ITaskKey[] = [1, '1', 0, '', a, b, NaN] + const doTask = vi.fn((task: { key: ITaskKey, value: number }) => task.value) + const schedule = new AsyncTask({ doTask, getTaskKey: task => task.key, maxWaitingGap: 0 }) + const tasks = keys.map((key, value) => ({ key, value })) + const result = schedule.dispatch([...tasks, { key: a, value: 99 }, { key: NaN, value: 99 }]) + await vi.runAllTimersAsync() + await expect(result).resolves.toEqual([0, 1, 2, 3, 4, 5, 6, 4, 6]) + expect(doTask).toHaveBeenCalledTimes(keys.length) + }) + + it('supports custom comparison when no key extractor is supplied', async () => { + const doTask = vi.fn((task: { id: number, label: string }) => task.label) + const schedule = new AsyncTask({ doTask, isSameTask: (a, b) => a.id === b.id, maxWaitingGap: 0 }) + const result = schedule.dispatch([{ id: 1, label: 'first' }, { id: 1, label: 'second' }]) + await vi.runAllTimersAsync() + await expect(result).resolves.toEqual(['first', 'first']) + expect(doTask).toHaveBeenCalledTimes(1) + }) + + it('shares a keyed executor across hundreds of overlapping requests', async () => { + const doTask = vi.fn((n: number) => n * 10) + const schedule = new AsyncTask({ doTask, getTaskKey: n => n, maxWaitingGap: 0 }) + const requests = Array.from({ length: 200 }, (_, i) => [i % 20, (i + 1) % 20, i % 20]) + const responses = requests.map(tasks => schedule.dispatch(tasks)) + await vi.runAllTimersAsync() + expect(await Promise.all(responses)).toEqual(requests.map(tasks => tasks.map(n => n * 10))) + expect(doTask).toHaveBeenCalledTimes(20) + expect(schedule.isTaskRunning).toBe(false) + }) + + it('does linear key work for a large batch and one key lookup per hot-cache hit', async () => { + const batchDoTasks = vi.fn((tasks: number[]) => tasks) + const getTaskKey = vi.fn((n: number) => n) + const isSameTask = vi.fn(AsyncTask.isEqual) + const schedule = new AsyncTask({ batchDoTasks, getTaskKey, isSameTask, maxWaitingGap: 0, invalidAfter: 1000 }) + const tasks = Array.from({ length: 1000 }, (_, i) => i) + const result = schedule.dispatch([...tasks, ...tasks]) + await vi.runAllTimersAsync() + await expect(result).resolves.toEqual([...tasks, ...tasks]) + expect(getTaskKey.mock.calls.length).toBeLessThan(12 * tasks.length) + getTaskKey.mockClear() + const cached = await Promise.all(tasks.map(task => schedule.dispatch(task))) + expect(cached).toEqual(tasks) + expect(getTaskKey).toHaveBeenCalledTimes(tasks.length) + expect(isSameTask).not.toHaveBeenCalled() + expect(batchDoTasks).toHaveBeenCalledTimes(1) + }) + + it.each([0.5, 1.5])('expires fractional TTL %s at the next integer millisecond', async (invalidAfter) => { + const doTask = vi.fn((n: number) => n) + const schedule = new AsyncTask({ doTask, getTaskKey: n => n, maxWaitingGap: 0, invalidAfter }) + const first = schedule.dispatch(1) + await vi.runAllTimersAsync() + await first + await vi.advanceTimersByTimeAsync(Math.floor(invalidAfter) + 1) + const fresh = schedule.dispatch(1) + await vi.runAllTimersAsync() + await fresh + expect(doTask).toHaveBeenCalledTimes(2) + }) + + it('rejects a failing key extractor without leaving a queued request behind', async () => { + const error = new Error('invalid key') + const batchDoTasks = vi.fn((tasks: number[]) => tasks) + const schedule = new AsyncTask({ batchDoTasks, getTaskKey: n => { if (n === 2) throw error; return n }, maxWaitingGap: 0 }) + await expect(schedule.dispatch([1, 2])).rejects.toBe(error) + const next = schedule.dispatch([1]) + await vi.runAllTimersAsync() + await expect(next).resolves.toEqual([1]) + expect(batchDoTasks).toHaveBeenCalledTimes(1) + schedule.cleanCache() + const fresh = schedule.dispatch([1]) + await vi.runAllTimersAsync() + await fresh + expect(batchDoTasks).toHaveBeenCalledTimes(2) + }) + + it.each([false, true])('explicit undefined TTL keeps successes while discarding retryable errors, keyed=%s', async (keyed) => { + const error = new Error('failure') + const doTask = vi.fn((n: number): number => { if (n === 2) throw error; return n }) + const schedule = new AsyncTask({ doTask, invalidAfter: undefined, maxWaitingGap: 0, ...(keyed ? { getTaskKey: (n: number) => n } : {}) }) + const first = schedule.dispatch([1, 2]) + await vi.runAllTimersAsync() + await expect(first).resolves.toEqual([1, error]) + await vi.advanceTimersByTimeAsync(100_000) + await expect(schedule.dispatch(1)).resolves.toBe(1) + const retry = schedule.dispatch([1, 2]) + await vi.runAllTimersAsync() + await expect(retry).resolves.toEqual([1, error]) + expect(doTask.mock.calls.map(([n]) => n)).toEqual([1, 2, 2]) + }) +}) diff --git a/tsconfig.test.json b/tsconfig.test.json new file mode 100644 index 0000000..65fd41e --- /dev/null +++ b/tsconfig.test.json @@ -0,0 +1,9 @@ +{ + "extends": "./tsconfig.json", + "compilerOptions": { + "rootDir": ".", + "noEmit": true + }, + "include": ["src", "test"], + "exclude": ["dist", "coverage", "node_modules"] +} diff --git a/vite.config.ts b/vite.config.ts index f24daf6..81e78d8 100644 --- a/vite.config.ts +++ b/vite.config.ts @@ -22,7 +22,9 @@ export default defineConfig({ globals: true, environment: 'node', coverage: { - reporter: ['text', 'lcov'] + reporter: ['text', 'lcov'], + include: ['src/**/*.ts'], + thresholds: { statements: 100, branches: 100, functions: 100, lines: 100 } } // ...additional Vitest settings if needed... },