From 8efbdc03de169f6e15eefb4b9acc5810047d5071 Mon Sep 17 00:00:00 2001 From: Saiya Date: Sat, 3 Oct 2026 01:01:23 -0700 Subject: [PATCH 1/5] fix: correct task scheduling, errors, and package types --- .github/workflows/main.yml | 48 +++++----- package.json | 4 +- readme.md | 30 +++--- src/index.ts | 115 +++++++++++++---------- test/async-task-static-methods.spec.ts | 10 ++ test/async-task.spec.ts | 89 ++++++------------ test/scheduler-regressions.spec.ts | 122 +++++++++++++++++++++++++ 7 files changed, 269 insertions(+), 149 deletions(-) create mode 100644 test/scheduler-regressions.spec.ts diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml index 7f01ad5..d3f5108 100644 --- a/.github/workflows/main.yml +++ b/.github/workflows/main.yml @@ -1,36 +1,32 @@ -# 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 + - 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 + run: yarn test + - name: Build library + run: yarn build diff --git a/package.json b/package.json index 6a398bf..59d6d30 100644 --- a/package.json +++ b/package.json @@ -4,9 +4,10 @@ "description": "schedule async tasks", "main": "dist/index.umd.js", "module": "dist/index.es.js", - "types": "dist/types.d.ts", + "types": "dist/index.d.ts", "scripts": { "build": "vite build", + "typecheck": "tsc --noEmit", "test": "vitest run --coverage", "release": "yarn test && yarn build && yarn publish" }, @@ -22,6 +23,7 @@ "license": "MIT", "private": false, "source": "src/index.ts", + "files": ["dist"], "devDependencies": { "@types/node": "^20.0.0", "@vitest/coverage-v8": "2.1.5", diff --git a/readme.md b/readme.md index 8aa8380..fd0627f 100644 --- a/readme.md +++ b/readme.md @@ -85,7 +85,7 @@ interface ITaskScheduleOptions { * 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 + batchDoTasks?: (tasks: Task[]) => Promise> | Array /** * action to do single task, can be async or sync function @@ -102,33 +102,33 @@ interface ITaskScheduleOptions { /** * max task count for batchDoTasks, default unlimited - * undefined or 0 for unlimited + * undefined or 0 for unlimited; otherwise a positive integer */ maxBatchCount?: number /** * batch tasks executing strategy, default parallel - * only works if maxBatchCount is specified and tasks more than maxBatchCount are executed + * serial preserves maxBatchCount; doTask defaults to one task at a time * * 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' + 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' + taskWaitingStrategy?: 'throttle' | 'debounce' /** * task waiting time in milliseconds, default 50ms * differently according to taskWaitingStrategy */ - maxWaitingGap: number + maxWaitingGap?: number /** @@ -175,7 +175,7 @@ const result12 = await taskSchedule.dispatch(1) // 1 ### 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` +an empty task list resolves immediately to `[]`; this method won't throw task execution errors, it will fulfil even partially failed, you can check whether its success by `response instanceof Error` ```ts import TaskSchedule from 'async-task-schedule' @@ -191,14 +191,14 @@ const taskSchedule = new TaskSchedule({ 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 +const resultOf1 = result[0] // Error +// first result is error +const isError = result[0] instanceof Error // true try { // will throw an error - const result2 = await taskSchedule.dispatch(2) // 1 + const result2 = await taskSchedule.dispatch(1) // throws } catch(error) { console.warn(error) } @@ -220,9 +220,9 @@ const taskSchedule = new TaskSchedule({ invalidAfter: 0 }) -const result1 = await taskSchedule.dispatch(1) // 1 +const result1 = await taskSchedule.dispatch(2) // 4 try { - const result2 = await taskSchedule.dispatch(2), + const result2 = await taskSchedule.dispatch(1) } catch(error) { console.warn(error) } @@ -251,7 +251,7 @@ await Promise.all([ // clean all cached result taskSchedule.cleanCache() // task will execute again -const result = await taskSchedule.dispatch(1), +const result = await taskSchedule.dispatch(1) ``` @@ -293,7 +293,7 @@ const fetchSchedule = new TaskSchedule({ // 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') { + 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 diff --git a/src/index.ts b/src/index.ts index 9fccf1c..a799016 100644 --- a/src/index.ts +++ b/src/index.ts @@ -77,7 +77,7 @@ export default class AsyncTask { /** * batch tasks executing strategy, default parallel - * only works if maxBatchCount is specified and tasks more than maxBatchCount are executed + * serial preserves maxBatchCount; doTask defaults to one task at a time * * 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 @@ -117,9 +117,10 @@ export default class AsyncTask { */ private retryWhenFailed?: boolean - /** - * tasks ready to be executed - */ + /** Tasks currently being executed, used to deduplicate new requests. */ + private runningTasks: Task[] = [] + + /** Tasks ready to be executed. */ private pendingTasks: Task[] /** @@ -167,8 +168,9 @@ export default class AsyncTask { this.batchDoTasks = userOptions.batchDoTasks this.taskExecStrategy = userOptions.taskExecStrategy - if (this.taskExecStrategy === 'serial') { - this.maxBatchCount = 1 + if (this.maxBatchCount !== undefined && this.maxBatchCount !== 0 + && (!Number.isInteger(this.maxBatchCount) || this.maxBatchCount < 1)) { + throw new Error('maxBatchCount must be a positive integer or 0 for unlimited') } this.retryWhenFailed = userOptions.retryWhenFailed @@ -188,7 +190,8 @@ export default class AsyncTask { this.cleanupTasks() try { const result = this.tryGetTaskResult(tasks) - return Promise.resolve(result) + if (!Array.isArray(tasks) && result instanceof Error) return Promise.reject(result) + return result } catch (error) { // note all tasks are cached, just created new tasks } @@ -212,7 +215,7 @@ export default class AsyncTask { */ private cleanCacheIfNeeded() { if (!this.needCleanCache) return - if (this.pendingTasks.length || this.taskQueue.length) return + if (this.isTaskRunning || this.pendingTasks.length || this.taskQueue.length) return this.needCleanCache = false this.doneTaskMap = [] } @@ -238,6 +241,7 @@ export default class AsyncTask { if (this.pendingTasks.length) { myTasks = myTasks.filter((f) => !this.hasTask(this.pendingTasks, f)) } + myTasks = myTasks.filter((task) => !this.hasTask(this.runningTasks, task)) // remove done tasks if (myTasks.length) { myTasks = myTasks.filter((f) => !this.getTaskResult(f)) @@ -261,50 +265,58 @@ export default class AsyncTask { // whether task is running isTaskRunning = false - private runTasks() { + private async runTasks() { if (this.isTaskRunning || !this.pendingTasks.length) return this.isTaskRunning = true - if (this.batchDoTasks) { - this.runTaskWithBatchDoTasks() - } else { - this.runTasksWithDoTask() + try { + while (this.pendingTasks.length) { + if (this.taskExecStrategy === 'serial') { + const count = this.maxBatchCount || (this.batchDoTasks ? this.pendingTasks.length : 1) + const tasks = this.pendingTasks.splice(0, count) + this.runningTasks.push(...tasks) + await this.executeTasks(tasks) + } else { + const tasks = this.pendingTasks.splice(0) + this.runningTasks.push(...tasks) + const count = this.maxBatchCount || tasks.length + const batches: Array> = [] + for (let i = 0; i < tasks.length; i += count) { + batches.push(this.executeTasks(tasks.slice(i, i + count))) + } + await Promise.all(batches) + } + } + } finally { + this.isTaskRunning = false + this.cleanupTasks() } } - private async runTasksWithDoTask() { - const taskItems = this.pendingTasks.splice(0, this.maxBatchCount || this.pendingTasks.length) - taskItems.forEach(async (task) => { + private async executeTasks(tasks: Task[]) { + if (this.batchDoTasks) { try { - const result = await this.doTask!(task) - this.updateResultMap([task], [result]) + this.updateResultMap(tasks, await this.batchDoTasks(tasks)) } catch (error) { - this.updateResultMap([task], AsyncTask.wrapError(error)) + this.updateResultMap(tasks, AsyncTask.wrapError(error)) } - this.checkAllTasks() - if (this.pendingTasks.length) { - this.runTasksWithDoTask() - } else { - this.cleanupTasks() - this.isTaskRunning = false - } - }) + this.finishTasks(tasks) + } else { + await Promise.all(tasks.map(async (task) => { + try { + this.updateResultMap([task], [await this.doTask!(task)]) + } catch (error) { + this.updateResultMap([task], AsyncTask.wrapError(error)) + } + this.finishTasks([task]) + })) + } } - private async runTaskWithBatchDoTasks() { - const taskItems = this.pendingTasks.splice(0, this.maxBatchCount || this.pendingTasks.length) - try { - const result = await this.batchDoTasks!(taskItems) - this.updateResultMap(taskItems, result) - } catch (error) { - this.updateResultMap(taskItems, AsyncTask.wrapError(error)) - } + private finishTasks(tasks: Task[]) { + this.runningTasks = this.runningTasks.filter((task) => !this.hasTask(tasks, task)) + if (!this.runningTasks.length && !this.pendingTasks.length) this.isTaskRunning = false this.checkAllTasks() - if (this.pendingTasks.length) { - this.runTaskWithBatchDoTasks() - } else { - this.cleanupTasks() - this.isTaskRunning = false - } + this.cleanupTasks() } /** @@ -316,7 +328,11 @@ export default class AsyncTask { const result = this.tryGetTaskResult(taskItem.tasks) // eslint-disable-next-line no-param-reassign taskItem.isDone = true - taskItem.resolve(result) + if (!Array.isArray(taskItem.tasks) && result instanceof Error) { + taskItem.reject(result) + } else { + taskItem.resolve(result) + } } catch (error) { // not found } @@ -332,6 +348,7 @@ export default class AsyncTask { * @param defaultResult default result if not found */ private tryGetTaskResult(tasks: Task[] | Task): (Result | Error) | Array { + if (Array.isArray(tasks) && !tasks.length) return [] // no cached data and no default result provided if (!this.doneTaskMap.length) throw new Error('no done task') @@ -384,7 +401,7 @@ export default class AsyncTask { private cleanupTasks() { this.cleanCacheIfNeeded() // has unresolved tasks, unable to cleanup task - if (this.taskQueue.length) return + if (this.isTaskRunning || this.taskQueue.length) return // nothing to cleanup if (!this.doneTaskMap.length) return // no need to remove outdated or failed tasks @@ -442,16 +459,20 @@ export default class AsyncTask { // 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 (typeA !== 'object' || a === null || b === null) 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) + 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 && AsyncTask.isEqual(a.getTime(), b.getTime()) + } // 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 + if (Object.keys(a).some(k => !Object.prototype.hasOwnProperty.call(b, k) || !AsyncTask.isEqual(a[k], b[k]))) return false return true } } 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..518cce2 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)) @@ -129,72 +131,39 @@ 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, + taskExecStrategy: 'serial', maxWaitingGap: 0, }) - 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); - }) - 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', () => { @@ -217,7 +186,7 @@ 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) }) 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/) + }) +}) From f2e07d2c2e358c634a5fb075163fdbc2ef353ccd Mon Sep 17 00:00:00 2001 From: Saiya Date: Sat, 3 Oct 2026 03:25:35 -0700 Subject: [PATCH 2/5] perf: index task identities and expand behavioral coverage --- package.json | 3 +- readme.md | 45 +++++ scripts/benchmark.cjs | 43 +++++ src/index.ts | 167 ++++++++++-------- test/async-task.spec.ts | 13 +- test/scheduler-behavior.spec.ts | 293 ++++++++++++++++++++++++++++++++ test/scheduler-types.spec.ts | 29 ++++ test/task-identity.spec.ts | 114 +++++++++++++ tsconfig.test.json | 9 + vite.config.ts | 4 +- 10 files changed, 639 insertions(+), 81 deletions(-) create mode 100644 scripts/benchmark.cjs create mode 100644 test/scheduler-behavior.spec.ts create mode 100644 test/scheduler-types.spec.ts create mode 100644 test/task-identity.spec.ts create mode 100644 tsconfig.test.json diff --git a/package.json b/package.json index 59d6d30..985bba2 100644 --- a/package.json +++ b/package.json @@ -7,7 +7,8 @@ "types": "dist/index.d.ts", "scripts": { "build": "vite build", - "typecheck": "tsc --noEmit", + "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" }, diff --git a/readme.md b/readme.md index fd0627f..7ff1ebd 100644 --- a/readme.md +++ b/readme.md @@ -100,6 +100,13 @@ interface ITaskScheduleOptions { */ isSameTask?: (a: Task, b: Task) => boolean + /** + * optional stable task identity, used for indexed deduplication and cache lookup + * same keys mean the same task; takes precedence over isSameTask + * return a string, number, or symbol + */ + getTaskKey?: (task: Task) => string | number | symbol + /** * max task count for batchDoTasks, default unlimited * undefined or 0 for unlimited; otherwise a positive integer @@ -367,3 +374,41 @@ const taskSchedule = new TaskSchedule({ ``` + +## Performance and testing + +For many tasks or large object inputs, supply a stable, inexpensive `getTaskKey`: + +```ts +const users = new TaskSchedule({ + doTask: async (task: { tenantId: string; userId: string }) => { + return fetch(`/tenants/${task.tenantId}/users/${task.userId}`) + }, + 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. + +The indexed path avoids linear searches for pending, running, and cached tasks. +Numeric TTLs skip cache scans until the next expiration; function-based TTLs are +still evaluated at cleanup so callbacks retain their behavior. Cache cleanup is +lazy and deferred during active work, as before. + +Run `yarn test` for behavior tests and coverage, `yarn typecheck` for source and +test/type assertions, and `yarn benchmark` for a reproducible deep-comparison vs. +keyed benchmark. The benchmark verifies result equality and reports median cold +and hot-cache timings plus comparison counts. Timings depend on the machine; +unit tests assert bounded key/comparison work rather than elapsed-time targets. + +Further optimization opportunities include notifying only requests affected by a +completed task (the current queue is scanned on every completion), configurable +concurrency/backpressure, and bounded cache size. `maxBatchCount` is the batch size, +not a global concurrency limit in parallel mode. Unlimited caching can retain +results indefinitely; use a TTL or `cleanCache()` when appropriate. diff --git a/scripts/benchmark.cjs b/scripts/benchmark.cjs new file mode 100644 index 0000000..cabd7aa --- /dev/null +++ b/scripts/benchmark.cjs @@ -0,0 +1,43 @@ +const { performance } = require('node:perf_hooks') +const assert = require('node:assert/strict') +const AsyncTask = require('../dist/index.umd.js') + +async function sample(count, keyed) { + let comparisons = 0 + const tasks = Array.from({ length: count }, (_, id) => ({ id, scope: { tenant: 'demo' } })) + const schedule = new AsyncTask({ + batchDoTasks: tasks => tasks.map(task => task.id), + maxWaitingGap: 0, + invalidAfter: 0, + retryWhenFailed: false, + isSameTask: (a, b) => { comparisons++; return AsyncTask.isEqual(a, b) }, + ...(keyed ? { getTaskKey: task => task.id } : {}), + }) + const coldStart = performance.now() + const result = await schedule.dispatch([...tasks, ...tasks]) + const coldMs = performance.now() - coldStart + assert.deepEqual(result, [...tasks, ...tasks].map(task => task.id)) + const coldComparisons = comparisons + comparisons = 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)) + return { coldMs, hotMs, coldComparisons, hotComparisons: comparisons } +} + +async function main() { + const rows = [] + for (const count of [100, 500, 1000]) { + for (const keyed of [false, true]) { + await sample(count, keyed) + const runs = [] + for (let i = 0; i < 5; i++) runs.push(await sample(count, keyed)) + const median = field => runs.map(run => run[field]).sort((a, b) => a - b)[2] + rows.push({ tasks: count, identity: keyed ? 'getTaskKey' : 'deep comparison', cold_ms: +median('coldMs').toFixed(2), hot_ms: +median('hotMs').toFixed(2), cold_comparisons: median('coldComparisons'), hot_comparisons: median('hotComparisons') }) + } + } + console.log(`Node ${process.version}; median of five runs after warmup; milliseconds include scheduling overhead.`) + console.table(rows) +} +main().catch(error => { console.error(error); process.exitCode = 1 }) diff --git a/src/index.ts b/src/index.ts index a799016..37ee7e4 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,3 +1,4 @@ +export type ITaskKey = string | number | symbol export type ITaskExecStrategy = 'parallel' | 'serial' export type ITaskWaitingStrategy = 'throttle' | 'debounce' export interface IAsyncTaskOptions { @@ -46,6 +47,11 @@ export interface IAsyncTaskOptions { * check whether two tasks are identified the same */ isSameTask?: (a: Task, b: Task) => boolean + /** + * Optional stable task identity for indexed deduplication and cached lookups. + * Equal keys identify the same task; takes precedence over isSameTask. + */ + getTaskKey?: (task: Task) => ITaskKey } export default class AsyncTask { /** @@ -68,6 +74,10 @@ export default class AsyncTask { * default: AsyncTask.isEqual (deep comparison) */ private isSameTask: (a: Task, b: Task) => boolean + private getTaskKey?: (task: Task) => ITaskKey + private pendingKeys = new Set() + private runningKeys = new Set() + private resultByKey = new Map() /** * max task count for batchDoTasks, default unlimited @@ -128,7 +138,7 @@ export default class AsyncTask { * empty if all task are done */ private taskQueue: Array<{ - tasks: Task[]|Task, resolve: Function, reject: Function, isDone?: boolean }> + tasks: Task[]|Task, resolve: Function, reject: Function }> /** * cached task result @@ -139,6 +149,7 @@ export default class AsyncTask { * whether need to clean cache result, aka clean doneTaskMap */ private needCleanCache?: boolean + private nextCacheCleanup = Infinity /** * default task options */ @@ -157,6 +168,7 @@ export default class AsyncTask { this.doneTaskMap = [] this.taskQueue = [] this.isSameTask = userOptions.isSameTask + this.getTaskKey = userOptions.getTaskKey this.maxBatchCount = userOptions.maxBatchCount this.maxWaitingGap = userOptions.maxWaitingGap @@ -188,12 +200,10 @@ export default class AsyncTask { async dispatch(tasks: T): Promise<{ [k in keyof T]: Result | Error } > async dispatch(tasks: Task[] | Task) { this.cleanupTasks() - try { - const result = this.tryGetTaskResult(tasks) - if (!Array.isArray(tasks) && result instanceof Error) return Promise.reject(result) - return result - } catch (error) { - // note all tasks are cached, just created new tasks + const result = this.findTaskResults(tasks) + if (result) { + if (!Array.isArray(tasks) && result.value instanceof Error) throw result.value + return result.value } return new Promise((resolve, reject) => { this.createTasks(tasks, resolve, reject) @@ -218,6 +228,8 @@ export default class AsyncTask { if (this.isTaskRunning || this.pendingTasks.length || this.taskQueue.length) return this.needCleanCache = false this.doneTaskMap = [] + this.resultByKey.clear() + this.nextCacheCleanup = Infinity } /** tasks combine waiting timeout */ @@ -233,21 +245,25 @@ export default class AsyncTask { * @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)) - } - myTasks = myTasks.filter((task) => !this.hasTask(this.runningTasks, task)) - // remove done tasks - if (myTasks.length) { - myTasks = myTasks.filter((f) => !this.getTaskResult(f)) + const requested = Array.isArray(tasks) ? tasks : [tasks] + let myTasks: Task[] + if (this.getTaskKey) { + const seen = new Set() + myTasks = requested.filter((task) => { + const key = this.getTaskKey!(task) + if (seen.has(key) || this.pendingKeys.has(key) || this.runningKeys.has(key) || this.resultByKey.has(key)) return false + seen.add(key) + return true + }) + for (const task of myTasks) this.pendingKeys.add(this.getTaskKey(task)) + } else { + myTasks = requested.filter((task, idx) => idx === requested.findIndex(t => this.isSameTask(t, task))) + myTasks = myTasks.filter((task) => !this.hasTask(this.pendingTasks, task) + && !this.hasTask(this.runningTasks, task) && !this.getTaskResult(task)) } + this.taskQueue.push({ tasks, resolve, reject }) if (!myTasks.length) return - this.pendingTasks = this.pendingTasks.concat(myTasks) + for (const task of myTasks) this.pendingTasks.push(task) clearTimeout(this.timeoutId) let timeout = 0 @@ -273,11 +289,11 @@ export default class AsyncTask { if (this.taskExecStrategy === 'serial') { const count = this.maxBatchCount || (this.batchDoTasks ? this.pendingTasks.length : 1) const tasks = this.pendingTasks.splice(0, count) - this.runningTasks.push(...tasks) + this.markRunning(tasks) await this.executeTasks(tasks) } else { const tasks = this.pendingTasks.splice(0) - this.runningTasks.push(...tasks) + this.markRunning(tasks) const count = this.maxBatchCount || tasks.length const batches: Array> = [] for (let i = 0; i < tasks.length; i += count) { @@ -312,62 +328,62 @@ export default class AsyncTask { } } + private markRunning(tasks: Task[]) { + if (this.getTaskKey) { + for (const task of tasks) { + const key = this.getTaskKey(task) + this.pendingKeys.delete(key) + this.runningKeys.add(key) + } + } else { + for (const task of tasks) this.runningTasks.push(task) + } + } + private finishTasks(tasks: Task[]) { - this.runningTasks = this.runningTasks.filter((task) => !this.hasTask(tasks, task)) - if (!this.runningTasks.length && !this.pendingTasks.length) this.isTaskRunning = false + if (this.getTaskKey) { + for (const task of tasks) this.runningKeys.delete(this.getTaskKey(task)) + } else { + this.runningTasks = this.runningTasks.filter((task) => !this.hasTask(tasks, task)) + } + if (!this.runningTasks.length && !this.runningKeys.size && !this.pendingTasks.length) this.isTaskRunning = false this.checkAllTasks() this.cleanupTasks() } - /** - * check all tasks, try to resolve - */ + /** Resolve requests whose results are all available. */ private checkAllTasks() { - this.taskQueue.forEach((taskItem) => { - try { - const result = this.tryGetTaskResult(taskItem.tasks) - // eslint-disable-next-line no-param-reassign - taskItem.isDone = true - if (!Array.isArray(taskItem.tasks) && result instanceof Error) { - taskItem.reject(result) - } else { - taskItem.resolve(result) - } - } catch (error) { - // not found + this.taskQueue = this.taskQueue.filter((taskItem) => { + const result = this.findTaskResults(taskItem.tasks) + if (!result) return true + if (!Array.isArray(taskItem.tasks) && result.value instanceof Error) { + taskItem.reject(result.value) + } else { + taskItem.resolve(result.value) } + return false }) - // 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 { - if (Array.isArray(tasks) && !tasks.length) return [] - // no cached data and no default result provided - if (!this.doneTaskMap.length) throw new Error('no done task') - + /** A wrapper distinguishes a cached undefined result from a cache miss. */ + private findTaskResults(tasks: Task[] | Task): { value: Result | Error | Array } | undefined { 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 values: Array = [] + for (const task of tasks) { + const result = this.getTaskResult(task) + if (!result) return undefined + values.push(result[1]) + } + return { value: values } } - const val = this.getTaskResult(tasks) - if (!val) throw new Error('not found') - return val[1] + const result = this.getTaskResult(tasks) + return result ? { value: result[1] } : undefined } private getTaskResult(task: Task): [Task, Result | Error] | undefined { - const result = this.doneTaskMap.find((t) => this.isSameTask(task, t.task)) + const result = this.getTaskKey + ? this.resultByKey.get(this.getTaskKey(task)) + : this.doneTaskMap.find((t) => this.isSameTask(task, t.task)) if (result) { return [result.task, result.value] } @@ -380,17 +396,24 @@ export default class AsyncTask { private updateResultMap(tasks: Task[], result: Array | Error) { const now = Date.now() - let doneArray: any[] = [] + let doneArray: Array<{ task: Task, value: Result | Error, time: number }> = [] if (result instanceof Error) { doneArray = tasks.map((t) => ({ task: t, value: result, time: now })) } else { - const defaultValue = new Error('not found') + let defaultValue: Error | undefined doneArray = tasks.map((t, idx) => { - const taskResult = result.length > idx ? result[idx] : defaultValue + const taskResult = result.length > idx ? result[idx] : (defaultValue || (defaultValue = new Error('not found'))) return { task: t, value: taskResult, time: now } }) } - this.doneTaskMap = this.doneTaskMap.concat(doneArray) + for (const item of doneArray) { + this.doneTaskMap.push(item) + if (this.retryWhenFailed && item.value instanceof Error) this.nextCacheCleanup = 0 + if (typeof this.invalidAfter === 'number' && this.invalidAfter) { + this.nextCacheCleanup = Math.min(this.nextCacheCleanup, Math.floor(now + this.invalidAfter) + 1) + } + if (this.getTaskKey) this.resultByKey.set(this.getTaskKey(item.task), item) + } } /** @@ -407,6 +430,8 @@ export default class AsyncTask { // no need to remove outdated or failed tasks if (!this.invalidAfter && !this.retryWhenFailed) return const now = Date.now() + if (typeof this.invalidAfter !== 'function' && now < this.nextCacheCleanup) return + this.nextCacheCleanup = Infinity this.doneTaskMap = this.doneTaskMap.filter((item) => { if (this.retryWhenFailed && item.value instanceof Error) { return false @@ -414,10 +439,16 @@ export default class AsyncTask { 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 + const valid = now - item.time <= invalidAfter + if (valid) this.nextCacheCleanup = Math.min(this.nextCacheCleanup, Math.floor(item.time + invalidAfter) + 1) + return valid } return true }) + if (this.getTaskKey) { + this.resultByKey.clear() + for (const item of this.doneTaskMap) this.resultByKey.set(this.getTaskKey(item.task), item) + } } /** diff --git a/test/async-task.spec.ts b/test/async-task.spec.ts index 518cce2..4746e05 100644 --- a/test/async-task.spec.ts +++ b/test/async-task.spec.ts @@ -61,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') }) }) @@ -168,11 +163,7 @@ describe('async-task-schedule', () => { 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 () => { 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-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... }, From f0f736799fb97144e0901a65c1832fbb485fc0b0 Mon Sep 17 00:00:00 2001 From: Saiya Date: Sat, 3 Oct 2026 03:47:30 -0700 Subject: [PATCH 3/5] refactor: share task records and guard linear scheduling work --- .github/workflows/main.yml | 13 +- .gitignore | 4 +- package.json | 1 + readme.md | 52 ++- scripts/benchmark.cjs | 76 +++-- src/equal.ts | 28 ++ src/index.ts | 539 ++++++++++-------------------- test/async-task.spec.ts | 21 +- test/equality-boundaries.spec.ts | 33 ++ test/performance.spec.ts | 57 ++++ test/refactor-regressions.spec.ts | 230 +++++++++++++ 11 files changed, 633 insertions(+), 421 deletions(-) create mode 100644 src/equal.ts create mode 100644 test/equality-boundaries.spec.ts create mode 100644 test/performance.spec.ts create mode 100644 test/refactor-regressions.spec.ts diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml index d3f5108..f49454e 100644 --- a/.github/workflows/main.yml +++ b/.github/workflows/main.yml @@ -17,6 +17,8 @@ jobs: node: [22, 24] steps: - uses: actions/checkout@v4 + with: + fetch-depth: 0 - uses: actions/setup-node@v4 with: node-version: ${{ matrix.node }} @@ -26,7 +28,12 @@ jobs: run: yarn install --frozen-lockfile - name: Typecheck run: yarn typecheck - - name: Test + - name: Test including performance regression budgets run: yarn test - - name: Build library - run: yarn build + - name: Benchmark against the pre-refactor implementation + run: yarn benchmark + - 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/package.json b/package.json index 985bba2..cfec0e6 100644 --- a/package.json +++ b/package.json @@ -10,6 +10,7 @@ "benchmark": "npm run build && node scripts/benchmark.cjs", "typecheck": "tsc --noEmit && tsc --project tsconfig.test.json", "test": "vitest run --coverage", + "test:perf": "vitest run test/performance.spec.ts", "release": "yarn test && yarn build && yarn publish" }, "keywords": [ diff --git a/readme.md b/readme.md index 7ff1ebd..3f40897 100644 --- a/readme.md +++ b/readme.md @@ -396,19 +396,41 @@ A symbol must be reused, rather than created anew on each call. Keys use `Map` e `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-based TTLs are -still evaluated at cleanup so callbacks retain their behavior. Cache cleanup is -lazy and deferred during active work, as before. - -Run `yarn test` for behavior tests and coverage, `yarn typecheck` for source and -test/type assertions, and `yarn benchmark` for a reproducible deep-comparison vs. -keyed benchmark. The benchmark verifies result equality and reports median cold -and hot-cache timings plus comparison counts. Timings depend on the machine; -unit tests assert bounded key/comparison work rather than elapsed-time targets. - -Further optimization opportunities include notifying only requests affected by a -completed task (the current queue is scanned on every completion), configurable -concurrency/backpressure, and bounded cache size. `maxBatchCount` is the batch size, -not a global concurrency limit in parallel mode. Unlimited caching can retain -results indefinitely; use a TTL or `cleanCache()` when appropriate. +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 public API and batching strategies remain unchanged. diff --git a/scripts/benchmark.cjs b/scripts/benchmark.cjs index cabd7aa..e8cf05a 100644 --- a/scripts/benchmark.cjs +++ b/scripts/benchmark.cjs @@ -1,43 +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 AsyncTask = require('../dist/index.umd.js') +const ts = require('typescript') +const Current = require('../dist/index.umd.js') -async function sample(count, keyed) { - let comparisons = 0 +// 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 schedule = new AsyncTask({ - batchDoTasks: tasks => tasks.map(task => task.id), - maxWaitingGap: 0, - invalidAfter: 0, - retryWhenFailed: false, - isSameTask: (a, b) => { comparisons++; return AsyncTask.isEqual(a, b) }, - ...(keyed ? { getTaskKey: task => task.id } : {}), + 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() - const result = await schedule.dispatch([...tasks, ...tasks]) + 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.deepEqual(result, [...tasks, ...tasks].map(task => task.id)) - const coldComparisons = comparisons - comparisons = 0 + 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)) - return { coldMs, hotMs, coldComparisons, hotComparisons: comparisons } + assert.equal(executions, count, 'hot requests must use cache') + return { coldMs, hotMs, coldKeys, hotKeys: keyCalls } } async function main() { const rows = [] - for (const count of [100, 500, 1000]) { - for (const keyed of [false, true]) { - await sample(count, keyed) - const runs = [] - for (let i = 0; i < 5; i++) runs.push(await sample(count, keyed)) - const median = field => runs.map(run => run[field]).sort((a, b) => a - b)[2] - rows.push({ tasks: count, identity: keyed ? 'getTaskKey' : 'deep comparison', cold_ms: +median('coldMs').toFixed(2), hot_ms: +median('hotMs').toFixed(2), cold_comparisons: median('coldComparisons'), hot_comparisons: median('hotComparisons') }) + 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') }) + } } } - console.log(`Node ${process.version}; median of five runs after warmup; milliseconds include scheduling overhead.`) + 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/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 37ee7e4..692453c 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,3 +1,5 @@ +import { isEqual } from './equal' + export type ITaskKey = string | number | symbol export type ITaskExecStrategy = 'parallel' | 'serial' export type ITaskWaitingStrategy = 'throttle' | 'debounce' @@ -53,412 +55,235 @@ export interface IAsyncTaskOptions { */ getTaskKey?: (task: Task) => ITaskKey } -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 - private getTaskKey?: (task: Task) => ITaskKey - private pendingKeys = new Set() - private runningKeys = new Set() - private resultByKey = new Map() - - /** - * max task count for batchDoTasks, default unlimited - * undefined or 0 for unlimited - */ - private maxBatchCount?: number - - /** - * batch tasks executing strategy, default parallel - * serial preserves maxBatchCount; doTask defaults to one task at a time - * - * 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 currently being executed, used to deduplicate new requests. */ - private runningTasks: Task[] = [] - - /** Tasks ready to be executed. */ - private pendingTasks: Task[] +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 }> - - /** - * 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 - private nextCacheCleanup = Infinity - /** - * 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.getTaskKey = userOptions.getTaskKey - 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.maxBatchCount !== undefined && this.maxBatchCount !== 0 - && (!Number.isInteger(this.maxBatchCount) || 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.retryWhenFailed = userOptions.retryWhenFailed - this.invalidAfter = userOptions.invalidAfter - this.runTasks = this.runTasks.bind(this) + 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.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() - const result = this.findTaskResults(tasks) - if (result) { - if (!Array.isArray(tasks) && result.value instanceof Error) throw result.value - return result.value + 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.isTaskRunning || this.pendingTasks.length || this.taskQueue.length) return - this.needCleanCache = false - this.doneTaskMap = [] - this.resultByKey.clear() - this.nextCacheCleanup = Infinity + 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) { - const requested = Array.isArray(tasks) ? tasks : [tasks] - let myTasks: Task[] - if (this.getTaskKey) { - const seen = new Set() - myTasks = requested.filter((task) => { - const key = this.getTaskKey!(task) - if (seen.has(key) || this.pendingKeys.has(key) || this.runningKeys.has(key) || this.resultByKey.has(key)) return false - seen.add(key) - return true - }) - for (const task of myTasks) this.pendingKeys.add(this.getTaskKey(task)) - } else { - myTasks = requested.filter((task, idx) => idx === requested.findIndex(t => this.isSameTask(t, task))) - myTasks = myTasks.filter((task) => !this.hasTask(this.pendingTasks, task) - && !this.hasTask(this.runningTasks, task) && !this.getTaskResult(task)) + private clearIfIdle() { + if (this.clearRequested && !this.active && this.pendingHead === this.pending.length) { + this.records.clear() + this.nextExpiry = Infinity + this.clearRequested = false } - this.taskQueue.push({ tasks, resolve, reject }) - if (!myTasks.length) return - for (const task of myTasks) this.pendingTasks.push(task) + } - 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 async runTasks() { - if (this.isTaskRunning || !this.pendingTasks.length) return + /** 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 run() { + if (this.pumping || this.pendingHead === this.pending.length) return + this.pumping = true try { - while (this.pendingTasks.length) { - if (this.taskExecStrategy === 'serial') { - const count = this.maxBatchCount || (this.batchDoTasks ? this.pendingTasks.length : 1) - const tasks = this.pendingTasks.splice(0, count) - this.markRunning(tasks) - await this.executeTasks(tasks) + 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 tasks = this.pendingTasks.splice(0) - this.markRunning(tasks) - const count = this.maxBatchCount || tasks.length + const records = this.take(available) + const stride = this.options.maxBatchCount || records.length const batches: Array> = [] - for (let i = 0; i < tasks.length; i += count) { - batches.push(this.executeTasks(tasks.slice(i, i + count))) - } + for (let i = 0; i < records.length; i += stride) batches.push(this.execute(records.slice(i, i + stride))) await Promise.all(batches) } } } finally { - this.isTaskRunning = false - this.cleanupTasks() + this.pumping = false + this.isTaskRunning = this.active > 0 + this.clearIfIdle() } } - private async executeTasks(tasks: Task[]) { - if (this.batchDoTasks) { + private async execute(records: Array>) { + if (this.options.batchDoTasks) { + let results: Array try { - this.updateResultMap(tasks, await this.batchDoTasks(tasks)) + 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) { - this.updateResultMap(tasks, AsyncTask.wrapError(error)) + results = records.map(() => AsyncTask.wrapError(error)) } - this.finishTasks(tasks) + 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(tasks.map(async (task) => { - try { - this.updateResultMap([task], [await this.doTask!(task)]) - } catch (error) { - this.updateResultMap([task], AsyncTask.wrapError(error)) - } - this.finishTasks([task]) + 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 markRunning(tasks: Task[]) { - if (this.getTaskKey) { - for (const task of tasks) { - const key = this.getTaskKey(task) - this.pendingKeys.delete(key) - this.runningKeys.add(key) - } - } else { - for (const task of tasks) this.runningTasks.push(task) - } + private validity(record: TaskRecord) { + const ttl = this.options.invalidAfter + return typeof ttl === 'function' ? ttl.call(this, record.task, record.value!) : ttl } - private finishTasks(tasks: Task[]) { - if (this.getTaskKey) { - for (const task of tasks) this.runningKeys.delete(this.getTaskKey(task)) - } else { - this.runningTasks = this.runningTasks.filter((task) => !this.hasTask(tasks, task)) - } - if (!this.runningTasks.length && !this.runningKeys.size && !this.pendingTasks.length) this.isTaskRunning = false - this.checkAllTasks() - this.cleanupTasks() - } - - /** Resolve requests whose results are all available. */ - private checkAllTasks() { - this.taskQueue = this.taskQueue.filter((taskItem) => { - const result = this.findTaskResults(taskItem.tasks) - if (!result) return true - if (!Array.isArray(taskItem.tasks) && result.value instanceof Error) { - taskItem.reject(result.value) + 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 { - taskItem.resolve(result.value) + 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) } - return false - }) - } - - /** A wrapper distinguishes a cached undefined result from a cache miss. */ - private findTaskResults(tasks: Task[] | Task): { value: Result | Error | Array } | undefined { - if (Array.isArray(tasks)) { - const values: Array = [] - for (const task of tasks) { - const result = this.getTaskResult(task) - if (!result) return undefined - values.push(result[1]) - } - return { value: values } - } - const result = this.getTaskResult(tasks) - return result ? { value: result[1] } : undefined - } - - private getTaskResult(task: Task): [Task, Result | Error] | undefined { - const result = this.getTaskKey - ? this.resultByKey.get(this.getTaskKey(task)) - : this.doneTaskMap.find((t) => this.isSameTask(task, t.task)) - if (result) { - return [result.task, result.value] - } - } - - private hasTask(list: Task[], task: Task): boolean { - return list.some((item) => this.isSameTask(task, item)) - } - - - private updateResultMap(tasks: Task[], result: Array | Error) { - const now = Date.now() - let doneArray: Array<{ task: Task, value: Result | Error, time: number }> = [] - if (result instanceof Error) { - doneArray = tasks.map((t) => ({ task: t, value: result, time: now })) - } else { - let defaultValue: Error | undefined - doneArray = tasks.map((t, idx) => { - const taskResult = result.length > idx ? result[idx] : (defaultValue || (defaultValue = new Error('not found'))) - return { task: t, value: taskResult, time: now } - }) + } catch (error) { + // A policy failure settles this task and cannot escape the background pump. + value = AsyncTask.wrapError(error) + this.records.delete(record.key) } - for (const item of doneArray) { - this.doneTaskMap.push(item) - if (this.retryWhenFailed && item.value instanceof Error) this.nextCacheCleanup = 0 - if (typeof this.invalidAfter === 'number' && this.invalidAfter) { - this.nextCacheCleanup = Math.min(this.nextCacheCleanup, Math.floor(now + this.invalidAfter) + 1) - } - if (this.getTaskKey) this.resultByKey.set(this.getTaskKey(item.task), item) + record.value = value + record.state = 'done' + --this.active + if (!this.active && this.pendingHead === this.pending.length) { + this.isTaskRunning = false + this.pumping = false } + 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.isTaskRunning || 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() - if (typeof this.invalidAfter !== 'function' && now < this.nextCacheCleanup) return - this.nextCacheCleanup = Infinity - 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 - const valid = now - item.time <= invalidAfter - if (valid) this.nextCacheCleanup = Math.min(this.nextCacheCleanup, Math.floor(item.time + invalidAfter) + 1) - return valid - } - return true - }) - if (this.getTaskKey) { - this.resultByKey.clear() - for (const item of this.doneTaskMap) this.resultByKey.set(this.getTaskKey(item.task), item) + 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 } @@ -482,28 +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' || a === null || b === null) return false - // if one of them is regexp, check via regexp literal - 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 && AsyncTask.isEqual(a.getTime(), b.getTime()) - } - // 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 => !Object.prototype.hasOwnProperty.call(b, k) || !AsyncTask.isEqual(a[k], b[k]))) return false - return true + return isEqual(a, b) } } diff --git a/test/async-task.spec.ts b/test/async-task.spec.ts index 4746e05..56cad6f 100644 --- a/test/async-task.spec.ts +++ b/test/async-task.spec.ts @@ -182,22 +182,26 @@ describe('async-task-schedule', () => { 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 @@ -207,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) }) }) @@ -227,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) }) @@ -245,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) }) @@ -274,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]) +}) From 6b75688aed15911e983ac3d7e869cfe74170cbef Mon Sep 17 00:00:00 2001 From: Saiya Date: Sat, 3 Oct 2026 04:37:03 -0700 Subject: [PATCH 4/5] docs: clarify request batching and improve package discovery --- CHANGELOG.md | 23 ++ LICENSE | 21 ++ docs/performance.md | 67 ++++++ package.json | 23 +- readme.md | 543 +++++++++++++++----------------------------- 5 files changed, 309 insertions(+), 368 deletions(-) create mode 100644 CHANGELOG.md create mode 100644 LICENSE create mode 100644 docs/performance.md diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..b6b1d55 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,23 @@ +# 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. +- 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). 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/package.json b/package.json index cfec0e6..4ab3ccd 100644 --- a/package.json +++ b/package.json @@ -1,7 +1,7 @@ { "name": "async-task-schedule", "version": "1.0.1", - "description": "schedule async tasks", + "description": "Batch and deduplicate async requests with TTL caching.", "main": "dist/index.umd.js", "module": "dist/index.es.js", "types": "dist/index.d.ts", @@ -14,18 +14,25 @@ "release": "yarn test && yarn build && yarn publish" }, "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"], + "files": [ + "dist" + ], "devDependencies": { "@types/node": "^20.0.0", "@vitest/coverage-v8": "2.1.5", @@ -33,5 +40,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 3f40897..cae4800 100644 --- a/readme.md +++ b/readme.md @@ -1,436 +1,255 @@ # async-task-schedule +**Batch and deduplicate async requests with TTL caching.** - - -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 - - /** - * optional stable task identity, used for indexed deduplication and cache lookup - * same keys mean the same task; takes precedence over isSameTask - * return a string, number, or symbol - */ - getTaskKey?: (task: Task) => string | number | symbol - - /** - * max task count for batchDoTasks, default unlimited - * undefined or 0 for unlimited; otherwise a positive integer - */ - maxBatchCount?: number - - /** - * batch tasks executing strategy, default parallel - * serial preserves maxBatchCount; doTask defaults to one task at a time - * - * 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` -an empty task list resolves immediately to `[]`; this method won't throw task execution errors, 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] // Error -// first result is error -const isError = result[0] instanceof Error // true +| 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(1) // throws -} 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(2) // 4 -try { - const result2 = await taskSchedule.dispatch(1) -} 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. -attention: this action may not exec immediately, it will take effect after all tasks are done +### Limit batch size and execute batches serially + +**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 results = await squares.dispatch([2, -1, 3]) + console.log(results[0], results[1] instanceof Error, results[2]) // 4, true, 9 -const betterFetch = (resource: string, options?: RequestInit) => { - return fetchSchedule.dispatch({resource, options}) + 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()` -#### example 2: deal with `getUsers` -suppose we have a method `getUsers` defined as follows: +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. -```ts -// support max 5 users at a time -getUsers(userIds: string[]) => Promise<[{code: string, message: string, id?: string, name?: string, email?: string}]> -``` - -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)) -} +### Task identity -const getUserSchedule = new TaskSchedule({ - maxBatchCount: 5, - batchDoTasks: batchGetUsers, - // cache user info forever - invalidAfter: 0, -}) +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. -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']) -``` +**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`). -If you got a batch version function, you just need to make sure it throw an error when error occurred. +### Static utilities +- `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. -### 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 +## Performance and development -```ts -const taskSchedule = new TaskSchedule({ - ..., - taskExecStrategy: 'serial', - maxBatchCount: 2, -}) +```sh +yarn install --frozen-lockfile +yarn typecheck +yarn test +yarn test:perf +yarn benchmark ``` +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. +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. -## Performance and testing - -For many tasks or large object inputs, supply a stable, inexpensive `getTaskKey`: - -```ts -const users = new TaskSchedule({ - doTask: async (task: { tenantId: string; userId: string }) => { - return fetch(`/tenants/${task.tenantId}/users/${task.userId}`) - }, - getTaskKey: task => JSON.stringify([task.tenantId, task.userId]), -}) -``` +## License -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 public API and batching strategies remain unchanged. +[MIT](./LICENSE) From 10a199005b9b04017182d9a1aa2da39aca3c1cee Mon Sep 17 00:00:00 2001 From: Saiya Date: Sat, 3 Oct 2026 04:56:12 -0700 Subject: [PATCH 5/5] fix: preserve published package imports and document upgrade behavior --- .github/workflows/main.yml | 2 ++ CHANGELOG.md | 6 ++++ docs/upgrading.md | 50 ++++++++++++++++++++++++++ package.json | 9 ++--- readme.md | 5 +++ scripts/check-package.cjs | 73 ++++++++++++++++++++++++++++++++++++++ scripts/compat-package.cjs | 22 ++++++++++++ 7 files changed, 163 insertions(+), 4 deletions(-) create mode 100644 docs/upgrading.md create mode 100644 scripts/check-package.cjs create mode 100644 scripts/compat-package.cjs diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml index f49454e..f2bcc32 100644 --- a/.github/workflows/main.yml +++ b/.github/workflows/main.yml @@ -32,6 +32,8 @@ jobs: 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: diff --git a/CHANGELOG.md b/CHANGELOG.md index b6b1d55..866b8f8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,9 @@ - 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, @@ -21,3 +24,6 @@ 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/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 4ab3ccd..ec3d745 100644 --- a/package.json +++ b/package.json @@ -2,16 +2,17 @@ "name": "async-task-schedule", "version": "1.0.1", "description": "Batch and deduplicate async requests with TTL caching.", - "main": "dist/index.umd.js", + "main": "dist/index.js", "module": "dist/index.es.js", - "types": "dist/index.d.ts", + "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", "test:perf": "vitest run test/performance.spec.ts", - "release": "yarn test && yarn build && yarn publish" + "release": "yarn test && yarn build && yarn publish", + "test:package": "npm run build && node scripts/check-package.cjs" }, "keywords": [ "batch", diff --git a/readme.md b/readme.md index cae4800..155c7cf 100644 --- a/readme.md +++ b/readme.md @@ -237,6 +237,7 @@ yarn install --frozen-lockfile yarn typecheck yarn test yarn test:perf +yarn test:package yarn benchmark ``` @@ -249,6 +250,10 @@ See [performance and testing](https://github.com/oe/async-task-schedule/blob/mai 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 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`)