@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 <qxprakash@gmail.com>

.changeset/bumpy-cougars-open.md

@@ -0,0 +1,6 @@
+---
+"@uppy/xhr-upload": minor
+"@uppy/utils": minor
+---
+
+Introduce modern, minimal TaskQueue to replace RateLimitedQueue

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<void>()
+    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<string>()
+
+    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<string>()
+
+    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<void>()
+    const started2 = Promise.withResolvers<void>()
+    const blocker1 = Promise.withResolvers<void>()
+    const blocker2 = Promise.withResolvers<void>()
+
+    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<void>()
+    const started2 = Promise.withResolvers<void>()
+    const started3 = Promise.withResolvers<void>()
+    const blocker1 = Promise.withResolvers<void>()
+    const blocker2 = Promise.withResolvers<void>()
+    const blocker3 = Promise.withResolvers<void>()
+
+    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<void>()
+    const started2 = Promise.withResolvers<void>()
+    const blocker1 = Promise.withResolvers<void>()
+
+    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)
+  })
+})

packages/@uppy/utils/src/TaskQueue.ts

@@ -0,0 +1,260 @@
+/**
+ * A promise that can be aborted.
+ */
+export interface AbortablePromise<T> extends Promise<T> {
+  abort(reason?: unknown): void
+  /**
+   * @deprecated Legacy compatibility - abort when signal fires
+   */
+  abortOn(signal?: AbortSignal): AbortablePromise<T>
+}
+
+interface QueuedTask<T> {
+  run: () => Promise<T>
+  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<unknown>[] = []
+  #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<T>(task: (signal: AbortSignal) => Promise<T>): AbortablePromise<T> {
+    const controller = new AbortController()
+
+    let resolve!: (value: T) => void
+    let reject!: (reason: unknown) => void
+
+    const promise = new Promise<T>((res, rej) => {
+      resolve = res
+      reject = rej
+    }) as AbortablePromise<T>
+
+    const queuedTask: QueuedTask<T> = {
+      run: () => task(controller.signal),
+      resolve,
+      reject,
+      controller,
+    }
+
+    // Handle abort while queued
+    controller.signal.addEventListener(
+      'abort',
+      () => {
+        const index = this.#queue.indexOf(queuedTask as QueuedTask<unknown>)
+        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<unknown>)
+    }
+
+    return promise
+  }
+
+  #execute<T>(task: QueuedTask<T>): 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<T>
+    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<T extends (...args: any[]) => Promise<any>>(
+    fn: T,
+  ): (...args: Parameters<T>) => AbortablePromise<Awaited<ReturnType<T>>> {
+    return (...args: Parameters<T>) => {
+      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)
+      })
+    }
+  }
+}

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,

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<string, EventManager<M, B> | 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<M, B>) {
     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<M, B>[]) {
     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<M, B>),
-        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<M, B>),
+          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<M, B>) => {
@@ -496,21 +485,15 @@ export default class XHRUpload<
           }
           this.uppy.on('file-removed', removedHandler)
 
-          const uploadPromise = this.uppy
+          return this.uppy
             .getRequestClientForFile<RequestClient<M, B>>(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',