From 6b1abaa541a405d401e75a8d77d4e1d55f0661b1 Mon Sep 17 00:00:00 2001 From: Merlijn Vos Date: Thu, 5 Feb 2026 15:46:45 +0100 Subject: [PATCH] @uppy/utils,@uppy/xhr-upload: create and use new queue (#6105) Closes #4173, kind off ## Problem Current situation with `RateLimitedQueue` - Track concurrency: keep a running count, only dispatch up to limit, accept Infinity. - Priority enqueue: queued handlers sorted by priority; dequeued in that order. - Lifecycle bookkeeping: each job gets abort()/done() hooks; abort for running vs queued differs; decrement active count and advance queue via microtasks. - Requeue placeholders: supports a shouldBeRequeued marker so callers can hold/retry slots without running immediately. - Function wrappers: wrap sync or async fns to enforce the queue and return abortable handles. - Cancellation plumbing: provides abortOn(signal) to bind queued/running work to an AbortSignal; abortable promises carry abort(). - Pausing: pause(duration?) freezes dispatch (optional auto-resume); resume() restarts up to the current limit. - Rate limiting/backoff: rateLimit(duration) pauses, drops concurrency, then ramps it back toward the previous upper bound over time. Case in point: it's some sort of made up, monstrosity data structure trying to be too many things. It also has rate limiting and exponential backoff but that's already in [`fetcher`](https://github.com/transloadit/uppy/blob/main/packages/%40uppy/utils/src/fetcher.ts) too. Would be better if we had separation of concerns and proven data structures. ## Solution A "dumb" promise-based priority queue that doesn't care at all about what a promise does or if it needs to be retried. The promise inside determines if it has retrying and exponential backoff, such as a `fetcher` promise. To not make this a breaking change and a huge diff, we still implement this per uploader, starting with xhr-upload, and keep backwards compatibility in the interface so we can still pass it to `companion-client`, which needs to share the same queue. ## Ideal future When you think about it, it's odd that we implement a queue per uploader if a queue is so central to uppy working correctly. It's even more odd that we also have to inject that queue into `companion-client` per uploader for queue sharing. Ideally, `core` is responsible for having the queue. All uploaders would do is push a promise to `core` and `core` doesn't care if that promise is a tus, xhr, or S3 upload. --------- Co-authored-by: Prakash --- .changeset/bumpy-cougars-open.md | 6 + packages/@uppy/utils/src/TaskQueue.test.ts | 254 ++++++++++++++++++++ packages/@uppy/utils/src/TaskQueue.ts | 260 +++++++++++++++++++++ packages/@uppy/utils/src/index.ts | 6 +- packages/@uppy/xhr-upload/src/index.ts | 119 ++++------ 5 files changed, 575 insertions(+), 70 deletions(-) create mode 100644 .changeset/bumpy-cougars-open.md create mode 100644 packages/@uppy/utils/src/TaskQueue.test.ts create mode 100644 packages/@uppy/utils/src/TaskQueue.ts diff --git a/.changeset/bumpy-cougars-open.md b/.changeset/bumpy-cougars-open.md new file mode 100644 index 000000000..ba7f4c611 --- /dev/null +++ b/.changeset/bumpy-cougars-open.md @@ -0,0 +1,6 @@ +--- +"@uppy/xhr-upload": minor +"@uppy/utils": minor +--- + +Introduce modern, minimal TaskQueue to replace RateLimitedQueue diff --git a/packages/@uppy/utils/src/TaskQueue.test.ts b/packages/@uppy/utils/src/TaskQueue.test.ts new file mode 100644 index 000000000..27c69fd71 --- /dev/null +++ b/packages/@uppy/utils/src/TaskQueue.test.ts @@ -0,0 +1,254 @@ +import { describe, expect, it } from 'vitest' +import delay from './delay.js' +import { TaskQueue } from './TaskQueue.js' + +describe('TaskQueue', () => { + it('runs queued tasks in FIFO order after resume', async () => { + const queue = new TaskQueue({ concurrency: 1 }) + queue.pause() + + const order: string[] = [] + const makeTask = (label: string) => async () => { + order.push(label) + await delay(5) + return label + } + + const first = queue.add(makeTask('first')) + const second = queue.add(makeTask('second')) + const third = queue.add(makeTask('third')) + + expect(queue.pending).toBe(3) + + queue.resume() + await Promise.all([first, second, third]) + + expect(order).toEqual(['first', 'second', 'third']) + }) + + it('aborts a queued task without executing it', async () => { + const queue = new TaskQueue({ concurrency: 1 }) + const runningBlocker = Promise.withResolvers() + const started: string[] = [] + + const running = queue.add(async () => { + started.push('first') + await runningBlocker.promise + }) + const queued = queue.add(async () => { + started.push('second') + return 'second' + }) + + const reason = new Error('nope') + queued.abort(reason) + + await expect(queued).rejects.toBe(reason) + + runningBlocker.resolve() + await running + + expect(started).toEqual(['first']) + expect(queue.pending).toBe(0) + }) + + it('rejects a running task when aborted before it resolves', async () => { + const queue = new TaskQueue({ concurrency: 1 }) + const deferred = Promise.withResolvers() + + const promise = queue.add(async () => deferred.promise) + const reason = new Error('stop') + promise.abort(reason) + + deferred.resolve('ok') + + await expect(promise).rejects.toBe(reason) + }) + + it('clear rejects pending tasks but leaves running tasks alone', async () => { + const queue = new TaskQueue({ concurrency: 1 }) + const runningDeferred = Promise.withResolvers() + + const running = queue.add(async () => runningDeferred.promise) + const queued1 = queue.add(async () => 'queued1') + const queued2 = queue.add(async () => 'queued2') + + const reason = new Error('cleared') + queue.clear(reason) + + await expect(queued1).rejects.toBe(reason) + await expect(queued2).rejects.toBe(reason) + + runningDeferred.resolve('ok') + await expect(running).resolves.toBe('ok') + expect(queue.pending).toBe(0) + }) + + it('wrapPromiseFunction queues work and preserves arguments', async () => { + const queue = new TaskQueue({ concurrency: 1 }) + queue.pause() + + const order: string[] = [] + const makeTask = (label: string) => async () => { + order.push(label) + return label + } + + const wrapped = queue.wrapPromiseFunction(async (value: string) => { + order.push(`wrapped:${value}`) + return value.toUpperCase() + }) + + const first = queue.add(makeTask('first')) + const wrappedPromise = wrapped('hello') + + expect(typeof wrappedPromise.abort).toBe('function') + + queue.resume() + + await expect(first).resolves.toBe('first') + await expect(wrappedPromise).resolves.toBe('HELLO') + expect(order).toEqual(['first', 'wrapped:hello']) + }) + + it('updates concurrency via setter and starts additional tasks', async () => { + const queue = new TaskQueue({ concurrency: 1 }) + const started1 = Promise.withResolvers() + const started2 = Promise.withResolvers() + const blocker1 = Promise.withResolvers() + const blocker2 = Promise.withResolvers() + + const first = queue.add(async () => { + started1.resolve() + await blocker1.promise + return 'first' + }) + + const second = queue.add(async () => { + started2.resolve() + await blocker2.promise + return 'second' + }) + + await started1.promise + expect(queue.running).toBe(1) + expect(queue.pending).toBe(1) + + queue.concurrency = 2 + await started2.promise + + expect(queue.running).toBe(2) + expect(queue.pending).toBe(0) + + blocker1.resolve() + blocker2.resolve() + await expect(first).resolves.toBe('first') + await expect(second).resolves.toBe('second') + }) + + it('aborts when abortOn signal fires while queued', async () => { + const queue = new TaskQueue({ concurrency: 1 }) + queue.pause() + + const controller = new AbortController() + const promise = queue.add(async () => 'ok') + const returned = promise.abortOn(controller.signal) + + expect(returned).toBe(promise) + + const reason = new Error('signal abort') + controller.abort(reason) + + await expect(promise).rejects.toBe(reason) + expect(queue.pending).toBe(0) + }) + + it('runs tasks concurrently up to the concurrency limit', async () => { + const queue = new TaskQueue({ concurrency: 2 }) + const started1 = Promise.withResolvers() + const started2 = Promise.withResolvers() + const started3 = Promise.withResolvers() + const blocker1 = Promise.withResolvers() + const blocker2 = Promise.withResolvers() + const blocker3 = Promise.withResolvers() + + const first = queue.add(async () => { + started1.resolve() + await blocker1.promise + return 'first' + }) + const second = queue.add(async () => { + started2.resolve() + await blocker2.promise + return 'second' + }) + const third = queue.add(async () => { + started3.resolve() + await blocker3.promise + return 'third' + }) + + await Promise.all([started1.promise, started2.promise]) + + let thirdStarted = false + started3.promise.then(() => { + thirdStarted = true + }) + + await delay(1) + expect(thirdStarted).toBe(false) + expect(queue.running).toBe(2) + expect(queue.pending).toBe(1) + + blocker1.resolve() + await started3.promise + + blocker2.resolve() + blocker3.resolve() + await expect(first).resolves.toBe('first') + await expect(second).resolves.toBe('second') + await expect(third).resolves.toBe('third') + }) + + it('continues processing after aborting a running task', async () => { + const queue = new TaskQueue({ concurrency: 1 }) + const started1 = Promise.withResolvers() + const started2 = Promise.withResolvers() + const blocker1 = Promise.withResolvers() + + const first = queue.add(async () => { + started1.resolve() + await blocker1.promise + return 'first' + }) + + const second = queue.add(async () => { + started2.resolve() + return 'second' + }) + + await started1.promise + const reason = new Error('abort running') + first.abort(reason) + blocker1.resolve() + + await expect(first).rejects.toBe(reason) + await started2.promise + await expect(second).resolves.toBe('second') + }) + + it('continues processing after a task throws synchronously', async () => { + const queue = new TaskQueue({ concurrency: 1 }) + const reason = new Error('sync throw') + + const first = queue.add(() => { + throw reason + }) + const second = queue.add(async () => 'second') + + await expect(first).rejects.toBe(reason) + await expect(second).resolves.toBe('second') + expect(queue.running).toBe(0) + expect(queue.pending).toBe(0) + }) +}) diff --git a/packages/@uppy/utils/src/TaskQueue.ts b/packages/@uppy/utils/src/TaskQueue.ts new file mode 100644 index 000000000..47259be2c --- /dev/null +++ b/packages/@uppy/utils/src/TaskQueue.ts @@ -0,0 +1,260 @@ +/** + * A promise that can be aborted. + */ +export interface AbortablePromise extends Promise { + abort(reason?: unknown): void + /** + * @deprecated Legacy compatibility - abort when signal fires + */ + abortOn(signal?: AbortSignal): AbortablePromise +} + +interface QueuedTask { + run: () => Promise + resolve: (value: T) => void + reject: (reason: unknown) => void + controller: AbortController +} + +export interface TaskQueueOptions { + concurrency?: number +} + +/** + * A concurrent task queue with FIFO ordering. + * + * Tasks are functions that receive an AbortSignal and return a Promise. + * The queue manages concurrency and processes tasks in insertion order. + * + * @example + * ```ts + * const queue = new TaskQueue({ concurrency: 3 }) + * + * const promise = queue.add(async (signal) => { + * const response = await fetch(url, { signal }) + * return response.json() + * }) + * + * // To abort: + * promise.abort() + * ``` + */ +export class TaskQueue { + #queue: QueuedTask[] = [] + #running = 0 + #concurrency: number + #paused = false + + constructor(options?: TaskQueueOptions) { + const limit = options?.concurrency + this.#concurrency = + typeof limit !== 'number' || limit === 0 ? Infinity : limit + } + + /** + * Add a task to the queue. + * + * @param task - Function receiving AbortSignal, returns Promise + * @returns AbortablePromise that resolves with task result + */ + add(task: (signal: AbortSignal) => Promise): AbortablePromise { + const controller = new AbortController() + + let resolve!: (value: T) => void + let reject!: (reason: unknown) => void + + const promise = new Promise((res, rej) => { + resolve = res + reject = rej + }) as AbortablePromise + + const queuedTask: QueuedTask = { + run: () => task(controller.signal), + resolve, + reject, + controller, + } + + // Handle abort while queued + controller.signal.addEventListener( + 'abort', + () => { + const index = this.#queue.indexOf(queuedTask as QueuedTask) + if (index !== -1) { + this.#queue.splice(index, 1) + reject( + controller.signal.reason ?? + new DOMException('Aborted', 'AbortError'), + ) + } + }, + { once: true }, + ) + + promise.abort = (reason?: unknown) => { + controller.abort(reason ?? new DOMException('Aborted', 'AbortError')) + } + + // Legacy compatibility: abortOn method + promise.abortOn = (signal?: AbortSignal) => { + if (signal) { + const onAbort = () => promise.abort(signal.reason) + signal.addEventListener('abort', onAbort, { once: true }) + promise.then( + () => signal.removeEventListener('abort', onAbort), + () => signal.removeEventListener('abort', onAbort), + ) + } + return promise + } + + // Run immediately or queue + if (!this.#paused && this.#running < this.#concurrency) { + this.#execute(queuedTask) + } else { + this.#queue.push(queuedTask as QueuedTask) + } + + return promise + } + + #execute(task: QueuedTask): void { + this.#running++ + + // Check if already aborted before starting + if (task.controller.signal.aborted) { + this.#running-- + task.reject( + task.controller.signal.reason ?? + new DOMException('Aborted', 'AbortError'), + ) + this.#advance() + return + } + + let runPromise: Promise + try { + runPromise = task.run() + } catch (error) { + runPromise = Promise.reject(error) + } + + runPromise + .then( + (result) => { + if (task.controller.signal.aborted) { + task.reject( + task.controller.signal.reason ?? + new DOMException('Aborted', 'AbortError'), + ) + } else { + task.resolve(result) + } + }, + (error) => { + task.reject(error) + }, + ) + .finally(() => { + this.#running-- + this.#advance() + }) + } + + #advance(): void { + // Use microtask to allow batch aborts without starting new tasks + queueMicrotask(() => { + if (this.#paused || this.#running >= this.#concurrency) return + + while (this.#queue.length > 0) { + const next = this.#queue.shift()! + if (next.controller.signal.aborted) continue + this.#execute(next) + return + } + }) + } + + /** + * Pause the queue. Running tasks continue, but no new tasks start. + */ + pause(): void { + this.#paused = true + } + + /** + * Resume the queue and start processing pending tasks. + */ + resume(): void { + this.#paused = false + // Kick off tasks up to concurrency limit + const available = this.#concurrency - this.#running + for (let i = 0; i < available; i++) { + this.#advance() + } + } + + /** + * Clear all pending tasks from the queue. + * Running tasks are not affected. + * + * @param reason - Optional reason for rejection (defaults to AbortError) + */ + clear(reason?: unknown): void { + const tasks = this.#queue.splice(0) + const error = reason ?? new DOMException('Cleared', 'AbortError') + for (const task of tasks) { + task.controller.abort(error) + task.reject(error) + } + } + + get concurrency(): number { + return this.#concurrency + } + + set concurrency(value: number) { + this.#concurrency = + typeof value !== 'number' || value === 0 ? Infinity : value + // If concurrency increased, try to start more tasks + if (!this.#paused) { + const available = this.#concurrency - this.#running + for (let i = 0; i < available; i++) { + this.#advance() + } + } + } + + get pending(): number { + return this.#queue.length + } + + get running(): number { + return this.#running + } + + get isPaused(): boolean { + return this.#paused + } + + /** + * @deprecated Legacy compatibility wrapper for RateLimitedQueue API. + * Wraps a function so that when called, it's queued and returns an AbortablePromise. + * Note: for legacy compatibility with RateLimitedQueue, the wrapped function + * does not receive this queue's AbortSignal. Aborting the returned promise + * will reject it, but it will not automatically cancel work inside the wrapped + * function unless that function is wired to an external AbortSignal. + */ + wrapPromiseFunction Promise>( + fn: T, + ): (...args: Parameters) => AbortablePromise>> { + return (...args: Parameters) => { + return this.add((signal) => { + // The wrapped function doesn't receive signal directly, + // caller is responsible for using signal if needed + void signal + return fn(...args) + }) + } + } +} diff --git a/packages/@uppy/utils/src/index.ts b/packages/@uppy/utils/src/index.ts index 467a89296..1b663a7ef 100644 --- a/packages/@uppy/utils/src/index.ts +++ b/packages/@uppy/utils/src/index.ts @@ -79,9 +79,13 @@ export { RateLimitedQueue, type WrapPromiseFunctionType, } from './RateLimitedQueue.js' - export { default as remoteFileObjToLocal } from './remoteFileObjToLocal.js' export { default as secondsToTime } from './secondsToTime.js' +export { + type AbortablePromise as TaskQueueAbortablePromise, + TaskQueue, + type TaskQueueOptions, +} from './TaskQueue.js' export type { I18n, Locale, diff --git a/packages/@uppy/xhr-upload/src/index.ts b/packages/@uppy/xhr-upload/src/index.ts index d41d1e9f3..46058ed60 100644 --- a/packages/@uppy/xhr-upload/src/index.ts +++ b/packages/@uppy/xhr-upload/src/index.ts @@ -15,12 +15,11 @@ import { filterFilesToEmitUploadStarted, filterFilesToUpload, getAllowedMetaFields, - internalRateLimitedQueue, isNetworkError, type LocalUppyFile, NetworkError, - RateLimitedQueue, type RemoteUppyFile, + TaskQueue, } from '@uppy/utils' import packageJson from '../package.json' with { type: 'json' } import locale from './locale.js' @@ -163,7 +162,7 @@ export default class XHRUpload< #getFetcher - requests: RateLimitedQueue + #queue: TaskQueue uploaderEvents: Record | null> @@ -180,13 +179,7 @@ export default class XHRUpload< this.i18nInit() - // Simultaneous upload limiting is shared across all uploads with this plugin. - if (internalRateLimitedQueue in this.opts) { - // @ts-ignore untyped internal - this.requests = this.opts[internalRateLimitedQueue] - } else { - this.requests = new RateLimitedQueue(this.opts.limit) - } + this.#queue = new TaskQueue({ concurrency: this.opts.limit }) if (this.opts.bundle && !this.opts.formData) { throw new Error( @@ -388,35 +381,32 @@ export default class XHRUpload< async #uploadLocalFile(file: LocalUppyFile) { const events = new EventManager(this.uppy) const controller = new AbortController() - const uppyFetch = this.requests.wrapPromiseFunction(async () => { - const opts = this.getOptions(file) - const fetch = this.#getFetcher([file]) - const body = opts.formData - ? this.createFormDataUpload(file, opts) - : file.data - const endpoint = - typeof opts.endpoint === 'string' - ? opts.endpoint - : await opts.endpoint(file) - return fetch(endpoint, { - ...opts, - body, - signal: controller.signal, - }) - }) events.onFileRemove(file.id, () => controller.abort()) - events.onCancelAll(file.id, () => { - controller.abort() - }) + events.onCancelAll(file.id, () => controller.abort()) try { - await uppyFetch() + await this.#queue.add(async (signal) => { + const opts = this.getOptions(file) + const fetch = this.#getFetcher([file]) + const body = opts.formData + ? this.createFormDataUpload(file, opts) + : file.data + const endpoint = + typeof opts.endpoint === 'string' + ? opts.endpoint + : await opts.endpoint(file) + return fetch(endpoint, { + ...opts, + body, + signal: AbortSignal.any([signal, controller.signal]), + }) + }) } catch (error) { - // TODO: create formal error with name 'AbortError' (this comes from RateLimitedQueue) - if (error.message !== 'Cancelled') { - throw error + if (error.name === 'AbortError') { + return } + throw error } finally { events.remove() } @@ -424,24 +414,6 @@ export default class XHRUpload< async #uploadBundle(files: LocalUppyFile[]) { const controller = new AbortController() - const uppyFetch = this.requests.wrapPromiseFunction(async () => { - const optsFromState = this.uppy.getState().xhrUpload ?? {} - const fetch = this.#getFetcher(files) - const body = this.createBundledUpload(files, { - ...this.opts, - ...optsFromState, - }) - const endpoint = - typeof this.opts.endpoint === 'string' - ? this.opts.endpoint - : await this.opts.endpoint(files) - return fetch(endpoint, { - // headers can't be a function with bundle: true - ...(this.opts as OptsWithHeaders), - body, - signal: controller.signal, - }) - }) function abort() { controller.abort() @@ -452,12 +424,29 @@ export default class XHRUpload< this.uppy.once('cancel-all', abort) try { - await uppyFetch() + await this.#queue.add(async (signal) => { + const optsFromState = this.uppy.getState().xhrUpload ?? {} + const fetch = this.#getFetcher(files) + const body = this.createBundledUpload(files, { + ...this.opts, + ...optsFromState, + }) + const endpoint = + typeof this.opts.endpoint === 'string' + ? this.opts.endpoint + : await this.opts.endpoint(files) + return fetch(endpoint, { + // headers can't be a function with bundle: true + ...(this.opts as OptsWithHeaders), + body, + signal: AbortSignal.any([signal, controller.signal]), + }) + }) } catch (error) { - // TODO: create formal error with name 'AbortError' (this comes from RateLimitedQueue) - if (error.message !== 'Cancelled') { - throw error + if (error.name === 'AbortError') { + return } + throw error } finally { this.uppy.off('cancel-all', abort) } @@ -488,7 +477,7 @@ export default class XHRUpload< await Promise.allSettled( files.map((file) => { if (file.isRemote) { - const getQueue = () => this.requests + const getQueue = () => this.#queue const controller = new AbortController() const removedHandler = (removedFile: UppyFile) => { @@ -496,21 +485,15 @@ export default class XHRUpload< } this.uppy.on('file-removed', removedHandler) - const uploadPromise = this.uppy + return this.uppy .getRequestClientForFile>(file) .uploadRemoteFile(file, this.#getCompanionClientArgs(file), { signal: controller.signal, getQueue, }) - - this.requests.wrapSyncFunction( - () => { + .finally(() => { this.uppy.off('file-removed', removedHandler) - }, - { priority: -1 }, - )() - - return uploadPromise + }) } return this.#uploadLocalFile(file) @@ -524,10 +507,8 @@ export default class XHRUpload< return } - // No limit configured by the user, and no RateLimitedQueue passed in by a "parent" plugin - // (basically just AwsS3) using the internal symbol - // @ts-ignore untyped internal - if (this.opts.limit === 0 && !this.opts[internalRateLimitedQueue]) { + // No limit configured by the user + if (this.opts.limit === 0) { this.uppy.log( '[XHRUpload] When uploading multiple files at once, consider setting the `limit` option (to `10` for example), to limit the number of concurrent uploads, which helps prevent memory and network issues: https://uppy.io/docs/xhr-upload/#limit-0', 'warning',