From 5012eb4124a2a4e490a4e68303f9a7b910e8c5f1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ren=C3=A9e=20Kooi?= Date: Mon, 13 Jul 2020 15:10:23 +0200 Subject: [PATCH] aws-s3-multipart: implement retries (#2312) --- CHANGELOG.md | 1 + examples/dev/Dashboard.js | 5 + package-lock.json | 6 + .../aws-s3-multipart/src/MultipartUploader.js | 143 ++++++++++++++---- packages/@uppy/aws-s3-multipart/src/index.js | 2 + .../@uppy/aws-s3-multipart/types/index.d.ts | 1 + packages/@uppy/utils/package.json | 1 + packages/@uppy/utils/src/AbortController.js | 20 +++ .../@uppy/utils/src/AbortController.test.js | 51 +++++++ packages/@uppy/utils/src/delay.js | 36 +++++ packages/@uppy/utils/src/delay.test.js | 33 ++++ website/src/docs/aws-s3-multipart.md | 24 +-- 12 files changed, 288 insertions(+), 35 deletions(-) create mode 100644 packages/@uppy/utils/src/AbortController.js create mode 100644 packages/@uppy/utils/src/AbortController.test.js create mode 100644 packages/@uppy/utils/src/delay.js create mode 100644 packages/@uppy/utils/src/delay.test.js diff --git a/CHANGELOG.md b/CHANGELOG.md index 3941bfa1a..862c26adf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -105,6 +105,7 @@ PRs are welcome! Please do open an issue to discuss first if it's a big feature, - [ ] core: add maxTotalFileSize restriction #514 (@arturi) - [ ] companion: what happens if access token expires during/between an download & upload (@ife) - [ ] providers: Provider Browser don't handle uppy restrictions, can we hide things that don't match the restrictions in Google Drive and Instagram? #1827 (@arturi) +- [x] aws-s3-multipart: retry uploading failed parts (#2312 / @goto-bus-stop) ## 1.16.1 diff --git a/examples/dev/Dashboard.js b/examples/dev/Dashboard.js index c0f959f84..3a6cc7bbe 100644 --- a/examples/dev/Dashboard.js +++ b/examples/dev/Dashboard.js @@ -11,6 +11,7 @@ const ScreenCapture = require('@uppy/screen-capture/src') const GoldenRetriever = require('@uppy/golden-retriever/src') const Tus = require('@uppy/tus/src') const AwsS3 = require('@uppy/aws-s3/src') +const AwsS3Multipart = require('@uppy/aws-s3-multipart/src') const XHRUpload = require('@uppy/xhr-upload/src') const Transloadit = require('@uppy/transloadit/src') const Form = require('@uppy/form/src') @@ -19,6 +20,7 @@ const Form = require('@uppy/form/src') const UPLOADER = 'tus' // const UPLOADER = 's3' +// const UPLOADER = 's3-multipart' // const UPLOADER = 'xhr' // const UPLOADER = 'transloadit' @@ -76,6 +78,9 @@ module.exports = () => { case 's3': uppyDashboard.use(AwsS3, { companionUrl: COMPANION_URL, limit: 6 }) break + case 's3-multipart': + uppyDashboard.use(AwsS3Multipart, { companionUrl: COMPANION_URL, limit: 6 }) + break case 'xhr': uppyDashboard.use(XHRUpload, { endpoint: XHR_ENDPOINT, limit: 6, bundle: true }) break diff --git a/package-lock.json b/package-lock.json index 4537c5343..5bd699b80 100644 --- a/package-lock.json +++ b/package-lock.json @@ -7786,6 +7786,7 @@ "@uppy/utils": { "version": "file:packages/@uppy/utils", "requires": { + "abortcontroller-polyfill": "^1.4.0", "lodash.throttle": "^4.1.1" } }, @@ -8973,6 +8974,11 @@ } } }, + "abortcontroller-polyfill": { + "version": "1.4.0", + "resolved": "https://registry.npmjs.org/abortcontroller-polyfill/-/abortcontroller-polyfill-1.4.0.tgz", + "integrity": "sha512-3ZFfCRfDzx3GFjO6RAkYx81lPGpUS20ISxux9gLxuKnqafNcFQo59+IoZqpO2WvQlyc287B62HDnDdNYRmlvWA==" + }, "absolute-path": { "version": "0.0.0", "resolved": "https://registry.npmjs.org/absolute-path/-/absolute-path-0.0.0.tgz", diff --git a/packages/@uppy/aws-s3-multipart/src/MultipartUploader.js b/packages/@uppy/aws-s3-multipart/src/MultipartUploader.js index 3c1056d39..d56270218 100644 --- a/packages/@uppy/aws-s3-multipart/src/MultipartUploader.js +++ b/packages/@uppy/aws-s3-multipart/src/MultipartUploader.js @@ -1,7 +1,11 @@ +const { AbortController, createAbortError } = require('@uppy/utils/lib/AbortController') +const delay = require('@uppy/utils/lib/delay') + const MB = 1024 * 1024 const defaultOptions = { limit: 1, + retryDelays: [0, 1000, 3000, 5000], getChunkSize (file) { return Math.ceil(file.size / 10000) }, @@ -14,11 +18,6 @@ const defaultOptions = { } } -function remove (arr, el) { - const i = arr.indexOf(el) - if (i !== -1) arr.splice(i, 1) -} - class MultipartUploader { constructor (file, options) { this.options = { @@ -31,6 +30,7 @@ class MultipartUploader { } this.file = file + this.abortController = new AbortController() this.key = this.options.key || null this.uploadId = this.options.uploadId || null @@ -45,15 +45,26 @@ class MultipartUploader { // aborting it immediately after it finishes. this.createdPromise = Promise.reject() // eslint-disable-line prefer-promise-reject-errors this.isPaused = false + this.partsInProgress = 0 this.chunks = null this.chunkState = null - this.uploading = [] this._initChunks() this.createdPromise.catch(() => {}) // silence uncaught rejection warning } + /** + * Was this upload aborted? + * + * If yes, we may need to throw an AbortError. + * + * @returns {boolean} + */ + _aborted () { + return this.abortController.signal.aborted + } + _initChunks () { const chunks = [] const desiredChunkSize = this.options.getChunkSize(this.file) @@ -79,6 +90,8 @@ class MultipartUploader { this.options.createMultipartUpload() ) return this.createdPromise.then((result) => { + if (this._aborted()) throw createAbortError() + const valid = typeof result === 'object' && result && typeof result.uploadId === 'string' && typeof result.key === 'string' @@ -103,6 +116,8 @@ class MultipartUploader { key: this.key }) ).then((parts) => { + if (this._aborted()) throw createAbortError() + parts.forEach((part) => { const i = part.PartNumber - 1 this.chunkState[i] = { @@ -128,7 +143,7 @@ class MultipartUploader { _uploadParts () { if (this.isPaused) return - const need = this.options.limit - this.uploading.length + const need = this.options.limit - this.partsInProgress if (need === 0) return // All parts are uploaded. @@ -149,7 +164,57 @@ class MultipartUploader { } candidates.forEach((index) => { - this._uploadPart(index) + this._uploadPartRetryable(index).catch((err) => { + this._onError(err) + }) + }) + } + + _retryable ({ before, attempt, after }) { + const { retryDelays } = this.options + const { signal } = this.abortController + + if (before) before() + + function shouldRetry (err) { + if (err.source && typeof err.source.status === 'number') { + const { status } = err.source + // 0 probably indicates network failure + return status === 0 || status === 409 || status === 423 || (status >= 500 && status < 600) + } + return false + } + + const doAttempt = (retryAttempt) => + attempt().catch((err) => { + if (this._aborted()) throw createAbortError() + + if (shouldRetry(err) && retryAttempt < retryDelays.length) { + return delay(retryDelays[retryAttempt], { signal }) + .then(() => doAttempt(retryAttempt + 1)) + } else { + throw err + } + }) + + return doAttempt(0).then((result) => { + if (after) after() + return result + }, (err) => { + if (after) after() + throw err + }) + } + + _uploadPartRetryable (index) { + return this._retryable({ + before: () => { + this.partsInProgress += 1 + }, + attempt: () => this._uploadPart(index), + after: () => { + this.partsInProgress -= 1 + } }) } @@ -170,11 +235,15 @@ class MultipartUploader { if (!valid) { throw new TypeError('AwsS3/Multipart: Got incorrect result from `prepareUploadPart()`, expected an object `{ url }`.') } + return result }).then(({ url, headers }) => { - this._uploadPartBytes(index, url, headers) - }, (err) => { - this._onError(err) + if (this._aborted()) { + this.chunkState[index].busy = false + throw createAbortError() + } + + return this._uploadPartBytes(index, url, headers) }) } @@ -202,6 +271,13 @@ class MultipartUploader { _uploadPartBytes (index, url, headers) { const body = this.chunks[index] + const { signal } = this.abortController + + let defer + const promise = new Promise((resolve, reject) => { + defer = { resolve, reject } + }) + const xhr = new XMLHttpRequest() xhr.open('PUT', url, true) if (headers) { @@ -211,7 +287,13 @@ class MultipartUploader { } xhr.responseType = 'text' - this.uploading.push(xhr) + function cleanup () { + signal.removeEventListener('abort', onabort) + } + function onabort () { + xhr.abort() + } + signal.addEventListener('abort', onabort) xhr.upload.addEventListener('progress', (ev) => { if (!ev.lengthComputable) return @@ -220,16 +302,20 @@ class MultipartUploader { }) xhr.addEventListener('abort', (ev) => { - remove(this.uploading, ev.target) + cleanup() this.chunkState[index].busy = false + + defer.reject(createAbortError()) }) xhr.addEventListener('load', (ev) => { - remove(this.uploading, ev.target) + cleanup() this.chunkState[index].busy = false if (ev.target.status < 200 || ev.target.status >= 300) { - this._onError(new Error('Non 2xx')) + const error = new Error('Non 2xx') + error.source = ev.target + defer.reject(error) return } @@ -238,23 +324,26 @@ class MultipartUploader { // NOTE This must be allowed by CORS. const etag = ev.target.getResponseHeader('ETag') if (etag === null) { - this._onError(new Error('AwsS3/Multipart: Could not read the ETag header. This likely means CORS is not configured correctly on the S3 Bucket. Seee https://uppy.io/docs/aws-s3-multipart#S3-Bucket-Configuration for instructions.')) + defer.reject(new Error('AwsS3/Multipart: Could not read the ETag header. This likely means CORS is not configured correctly on the S3 Bucket. Seee https://uppy.io/docs/aws-s3-multipart#S3-Bucket-Configuration for instructions.')) return } this._onPartComplete(index, etag) + defer.resolve() }) xhr.addEventListener('error', (ev) => { - remove(this.uploading, ev.target) + cleanup() this.chunkState[index].busy = false const error = new Error('Unknown error') error.source = ev.target - this._onError(error) + defer.reject(error) }) xhr.send(body) + + return promise } _completeUpload () { @@ -275,9 +364,8 @@ class MultipartUploader { } _abortUpload () { - this.uploading.slice().forEach(xhr => { - xhr.abort() - }) + this.abortController.abort() + this.createdPromise.then(() => { this.options.abortMultipartUpload({ key: this.key, @@ -286,10 +374,13 @@ class MultipartUploader { }, () => { // if the creation failed we do not need to abort }) - this.uploading = [] } _onError (err) { + if (err && err.name === 'AbortError') { + return + } + this.options.onError(err) } @@ -303,10 +394,10 @@ class MultipartUploader { } pause () { - const inProgress = this.uploading.slice() - inProgress.forEach((xhr) => { - xhr.abort() - }) + this.abortController.abort() + // Swap it out for a new controller, because this instance may be resumed later. + this.abortController = new AbortController() + this.isPaused = true } diff --git a/packages/@uppy/aws-s3-multipart/src/index.js b/packages/@uppy/aws-s3-multipart/src/index.js index b6b391c43..095f0ab67 100644 --- a/packages/@uppy/aws-s3-multipart/src/index.js +++ b/packages/@uppy/aws-s3-multipart/src/index.js @@ -28,6 +28,7 @@ module.exports = class AwsS3Multipart extends Plugin { const defaultOptions = { timeout: 30 * 1000, limit: 0, + retryDelays: [0, 1000, 3000, 5000], createMultipartUpload: this.createMultipartUpload.bind(this), listParts: this.listParts.bind(this), prepareUploadPart: this.prepareUploadPart.bind(this), @@ -198,6 +199,7 @@ module.exports = class AwsS3Multipart extends Plugin { onPartComplete, limit: this.opts.limit || 5, + retryDelays: this.opts.retryDelays || [], ...file.s3Multipart }) diff --git a/packages/@uppy/aws-s3-multipart/types/index.d.ts b/packages/@uppy/aws-s3-multipart/types/index.d.ts index 4425f63d8..2ad0839b0 100644 --- a/packages/@uppy/aws-s3-multipart/types/index.d.ts +++ b/packages/@uppy/aws-s3-multipart/types/index.d.ts @@ -33,6 +33,7 @@ declare module AwsS3Multipart { ) => MaybePromise<{ location?: string }> timeout?: number limit?: number + retryDelays?: number[] | null } } diff --git a/packages/@uppy/utils/package.json b/packages/@uppy/utils/package.json index 6ac8c1f2e..adcf5f0a0 100644 --- a/packages/@uppy/utils/package.json +++ b/packages/@uppy/utils/package.json @@ -18,6 +18,7 @@ "url": "git+https://github.com/transloadit/uppy.git" }, "dependencies": { + "abortcontroller-polyfill": "^1.4.0", "lodash.throttle": "^4.1.1" } } diff --git a/packages/@uppy/utils/src/AbortController.js b/packages/@uppy/utils/src/AbortController.js new file mode 100644 index 000000000..ee517c8f7 --- /dev/null +++ b/packages/@uppy/utils/src/AbortController.js @@ -0,0 +1,20 @@ +/** + * Little AbortController proxy module so we can swap out the implementation easily later. + */ + +const { AbortController, AbortSignal } = require('abortcontroller-polyfill/dist/abortcontroller') + +function createAbortError (message = 'Aborted') { + try { + return new DOMException(message, 'AbortError') + } catch { + // For Internet Explorer + const error = new Error(message) + error.name = 'AbortError' + return error + } +} + +exports.AbortController = AbortController +exports.AbortSignal = AbortSignal +exports.createAbortError = createAbortError diff --git a/packages/@uppy/utils/src/AbortController.test.js b/packages/@uppy/utils/src/AbortController.test.js new file mode 100644 index 000000000..64f795400 --- /dev/null +++ b/packages/@uppy/utils/src/AbortController.test.js @@ -0,0 +1,51 @@ +const { AbortController, AbortSignal } = require('./AbortController') + +function flushInstantTimeouts () { + return new Promise(resolve => setTimeout(resolve, 0)) +} + +describe('AbortController', () => { + it('has the expected shape', () => { + const controller = new AbortController() + expect(typeof controller.abort).toBe('function') + expect(controller.signal).toBeInstanceOf(AbortSignal) + }) + + it('emits "abort" when abort() is called', async () => { + const controller = new AbortController() + const callback = jest.fn() + + controller.signal.addEventListener('abort', callback) + controller.abort() + + await flushInstantTimeouts() + + expect(callback).toHaveBeenCalled() + expect(callback.mock.calls[0][0]).toBeInstanceOf(Event) + }) + + it('add and remove events', async () => { + const controller = new AbortController() + const callback = jest.fn() + const callback2 = jest.fn() + + controller.signal.addEventListener('abort', callback) + controller.signal.addEventListener('abort', callback2) + controller.signal.removeEventListener('abort', callback) + controller.abort() + + await flushInstantTimeouts() + + expect(callback2).toHaveBeenCalled() + expect(callback2.mock.calls[0][0]).toBeInstanceOf(Event) + expect(callback).not.toHaveBeenCalled() + }) + + it('sets `signal.aborted` property when abort() is called', () => { + const controller = new AbortController() + + expect(controller.signal.aborted).toBe(false) + controller.abort() + expect(controller.signal.aborted).toBe(true) + }) +}) diff --git a/packages/@uppy/utils/src/delay.js b/packages/@uppy/utils/src/delay.js new file mode 100644 index 000000000..0ecc781bc --- /dev/null +++ b/packages/@uppy/utils/src/delay.js @@ -0,0 +1,36 @@ +const { createAbortError } = require('./AbortController') + +/** + * Return a Promise that resolves after `ms` milliseconds. + * + * @param {number} ms - Number of milliseconds to wait. + * @param {{ signal?: AbortSignal }} [opts] - An abort signal that can be used to cancel the delay early. + * @returns {Promise} A Promise that resolves after the given amount of `ms`. + */ +module.exports = function delay (ms, opts) { + return new Promise((resolve, reject) => { + if (opts && opts.signal && opts.signal.aborted) { + return reject(createAbortError()) + } + + function onabort () { + clearTimeout(timeout) + cleanup() + reject(createAbortError()) + } + + const timeout = setTimeout(() => { + cleanup() + resolve() + }, ms) + + if (opts && opts.signal) { + opts.signal.addEventListener('abort', onabort) + } + function cleanup () { + if (opts && opts.signal) { + opts.signal.removeEventListener('abort', onabort) + } + } + }) +} diff --git a/packages/@uppy/utils/src/delay.test.js b/packages/@uppy/utils/src/delay.test.js new file mode 100644 index 000000000..1541adda0 --- /dev/null +++ b/packages/@uppy/utils/src/delay.test.js @@ -0,0 +1,33 @@ +const delay = require('./delay') +const { AbortController } = require('./AbortController') + +describe('delay', () => { + it('should wait for the specified time', async () => { + const start = Date.now() + await delay(100) + expect(Date.now() - start).toBeGreaterThanOrEqual(100) + }) + + it('should reject if signal is already aborted', async () => { + const signal = { aborted: true } + const start = Date.now() + await expect(delay(100, { signal })).rejects.toHaveProperty('name', 'AbortError') + // should really be instant but using a very large range in case CI decides to be super busy and block the event loop for a while + expect(Date.now() - start).toBeLessThan(50) + }) + + it('should reject when signal is aborted', async () => { + const controller = new AbortController() + const start = Date.now() + const testDelay = delay(100, { signal: controller.signal }) + await Promise.all([ + delay(50).then(() => controller.abort()), + expect(testDelay).rejects.toHaveProperty('name', 'AbortError') + ]) + + // should have rejected before the timer is done + const time = Date.now() - start + expect(time).toBeGreaterThanOrEqual(50) + expect(time).toBeLessThan(100) + }) +}) diff --git a/website/src/docs/aws-s3-multipart.md b/website/src/docs/aws-s3-multipart.md index e7255b2af..640e4cb06 100644 --- a/website/src/docs/aws-s3-multipart.md +++ b/website/src/docs/aws-s3-multipart.md @@ -38,23 +38,29 @@ const AwsS3Multipart = Uppy.AwsS3Multipart The `@uppy/aws-s3-multipart` plugin has the following configurable options: -### limit: 0 +### `limit: 0` The maximum amount of chunks to upload simultaneously. Set to `0` to disable limiting. -### companionUrl: null +### `retryDelays: [0, 1000, 3000, 5000]` + +When uploading a chunk fails, automatically try again after the millisecond intervals specified in this array. By default, we first retry instantly; if that fails, we retry after 1 second; if that fails, we retry after 3 seconds, etc. + +Set to `null` to disable automatic retries, and fail instantly if any chunk fails to upload. + +### `companionUrl: null` URL of the [Companion](/docs/companion) instance to use for proxying calls to the S3 Multipart API. This will be used by the default implementations of the upload-related functions below. If you provide your own implementations, a `companionUrl` is unnecessary. -### companionHeaders: {} +### `companionHeaders: {}` Custom headers that should be sent along to [Companion](/docs/companion) on every request. This will be used by the default implementations of the upload-related functions below. If you provide your own implementations, these headers are not sent automatically. -### getChunkSize(file) +### `getChunkSize(file)` A function that returns the minimum chunk size to use when uploading the given file. @@ -62,7 +68,7 @@ The S3 Multipart plugin uploads files in chunks. Each chunk requires a signing r S3 requires a minimum chunk size of 5MB, and supports at most 10,000 chunks per multipart upload. If `getChunkSize()` returns a size that's too small, Uppy will increase it to S3's minimum requirements. -### createMultipartUpload(file) +### `createMultipartUpload(file)` A function that calls the S3 Multipart API to create a new upload. `file` is the file object from Uppy's state. The most relevant keys are `file.name` and `file.type`. @@ -73,7 +79,7 @@ Return a Promise for an object with keys: The default implementation calls out to Companion's S3 signing endpoints. -### listParts(file, { uploadId, key }) +### `listParts(file, { uploadId, key })` A function that calls the S3 Multipart API to list the parts of a file that have already been uploaded. Receives the `file` object from Uppy's state, and an object with keys: @@ -88,7 +94,7 @@ Return a Promise for an array of S3 Part objects, as returned by the S3 Multipar The default implementation calls out to Companion's S3 signing endpoints. -### prepareUploadPart(file, partData) +### `prepareUploadPart(file, partData)` A function that generates a signed URL to upload a single part. Receives the `file` object from Uppy's state. The `partData` argument is an object with keys: @@ -113,7 +119,7 @@ Return a Promise for an object with keys: ``` - `headers` - **(Optional)** Custom headers that should be sent to the S3 presigned URL. -### abortMultipartUpload(file, { uploadId, key }) +### `abortMultipartUpload(file, { uploadId, key })` A function that calls the S3 Multipart API to abort a Multipart upload, and delete all parts that have been uploaded so far. Receives the `file` object from Uppy's state, and an object with keys: @@ -124,7 +130,7 @@ This is typically called when the user cancels an upload. Cancellation cannot fa The default implementation calls out to Companion's S3 signing endpoints. -### completeMultipartUpload(file, { uploadId, key, parts }) +### `completeMultipartUpload(file, { uploadId, key, parts })` A function that calls the S3 Multipart API to complete a Multipart upload, combining all parts into a single object in the S3 bucket. Receives the `file` object from Uppy's state, and an object with keys: