diff --git a/packages/@uppy/aws-s3/src/index.ts b/packages/@uppy/aws-s3/src/index.ts index c31ff2003..becbedeb8 100644 --- a/packages/@uppy/aws-s3/src/index.ts +++ b/packages/@uppy/aws-s3/src/index.ts @@ -7,9 +7,9 @@ import { } from '@uppy/core' import type { Body, Meta, UppyFile } from '@uppy/utils' import { - AbortController, filterFilesToEmitUploadStarted, filterFilesToUpload, + TaskQueue, } from '@uppy/utils' import packageJson from '../package.json' with { type: 'json' } import S3mini from './s3-client/S3.js' @@ -68,6 +68,16 @@ export interface AwsS3Options getChunkSize?: (file: { size: number }) => number allowedMetaFields?: string[] | boolean + /** + * Maximum number of files uploading concurrently. + * Each file uploads its parts sequentially. + * + * Default: 6 — chosen to match the browser's HTTP/1.1 per-origin connection + * limit. Most browsers allow 6 concurrent connections per host, so this + * prevents queueing at the browser level while maximizing throughput. + */ + limit?: number + /** * Custom function to generate the S3 object key. * Default: `{randomId}-{filename}` @@ -90,6 +100,8 @@ const MAX_PARTS = 10000 const defaultOptions = { shouldUseMultipart: (file: UppyFile) => (file.size || 0) > 100 * MB, allowedMetaFields: true, + // 6 matches browser HTTP/1.1 per-origin connection limit + limit: 6, } satisfies Partial> // ============================================================================ @@ -439,7 +451,13 @@ class S3Uploader { } #onSuccess(result: UploadResult): void { - // Clean up event listeners to prevent memory leaks + // If the upload was aborted (file removed mid-upload), the network request + // may still complete successfully. Don't emit success in this case since + // the file no longer exists in Uppy's state. + if (this.#abortController.signal.aborted) { + this.#eventManager.remove() + return + } this.#eventManager.remove() this.#options.onSuccess?.(result) } @@ -473,6 +491,7 @@ export default class AwsS3 extends BasePlugin< static VERSION = packageJson.version #s3Client!: S3mini + #queue!: TaskQueue #uploaders: Record | null> = {} constructor(uppy: Uppy, opts: AwsS3Options) { @@ -484,14 +503,16 @@ export default class AwsS3 extends BasePlugin< install(): void { this.#setResumableUploadsCapability(true) this.#initS3Client() + this.#queue = new TaskQueue({ concurrency: this.opts.limit }) this.uppy.addUploader(this.#upload) - this.uppy.on('cancel-all', this.#resetResumableCapability) + this.uppy.on('cancel-all', this.#handleCancelAll) } uninstall(): void { this.#setResumableUploadsCapability(false) this.uppy.removeUploader(this.#upload) - this.uppy.off('cancel-all', this.#resetResumableCapability) + this.uppy.off('cancel-all', this.#handleCancelAll) + this.#queue.clear() // Abort and clean up any in-flight uploads for (const fileId of Object.keys(this.#uploaders)) { const uploader = this.#uploaders[fileId] @@ -515,8 +536,9 @@ export default class AwsS3 extends BasePlugin< }) } - #resetResumableCapability = (): void => { + #handleCancelAll = (): void => { this.#setResumableUploadsCapability(true) + this.#queue.clear() } // -------------------------------------------------------------------------- @@ -609,7 +631,17 @@ export default class AwsS3 extends BasePlugin< new Error('Remote file uploads not yet supported'), ) } - return this.#uploadLocalFile(file) + return this.#queue.add(async () => { + // File may have been removed while waiting in the queue. + // Unlike actively uploading files, queued files don't have an S3Uploader + // instance yet, so there's no event listener to catch the removal. + // Re-fetch the file to ensure it still exists before starting upload. + const currentFile = this.uppy.getFile(file.id) + if (!currentFile) { + return + } + return this.#uploadLocalFile(currentFile) + }) }) await Promise.allSettled(promises)