diff --git a/src/lib/media/video/multipart/upload.ts b/src/lib/media/video/multipart/upload.ts index fb6cc7a5c3..6ca511f010 100644 --- a/src/lib/media/video/multipart/upload.ts +++ b/src/lib/media/video/multipart/upload.ts @@ -3,7 +3,7 @@ import {nanoid} from 'nanoid/non-secure' import {AbortError} from '#/lib/async/cancelable' import {type CompressedVideo} from '#/lib/media/video/types' -import {shouldRetryError} from '#/lib/strings/errors' +import {isRetryableHttpStatus, shouldRetryError} from '#/lib/strings/errors' import {getServiceAuthToken} from '../upload.shared' import {mimeToExt} from '../util' import { @@ -19,6 +19,7 @@ import {getMissingParts, planParts} from './planParts' import {createChunkReader} from './readChunk' import {createUploadPart} from './uploadPart' import {uploadParts} from './uploadParts' +import {delay} from './utils' export class MultipartFallbackError extends Error {} @@ -87,6 +88,7 @@ export async function uploadVideoMultipart({ ) } + // Preserve TypeScript's narrowing inside the recovery callback. const activeReader = reader if (!activeReader) throw new Error('Video chunk reader is unavailable') return await finishAndRecover({ @@ -165,8 +167,8 @@ async function finishAndRecover({ } return await abortThenFallbackOrResolve(jobId, token, finishError) case 'finishing': - // Finalization owns the reservation and may already have assembled - // the object. Retrying is idempotent; legacy fallback is unsafe. + // The service may have assembled the upload even though the finish + // request failed. Poll and retry instead of starting a second upload. await delay(1000, signal) continue case 'failed': @@ -215,7 +217,7 @@ function isRetryableStatusError(err: unknown) { (err instanceof MultipartUploadError && (err.error === 'ServiceOverloaded' || err.status === undefined || - err.status >= 500)) + isRetryableHttpStatus(err.status))) ) } @@ -293,17 +295,3 @@ async function getServiceAuthTokenWithRetry( function throwIfAborted(signal: AbortSignal) { if (signal.aborted) throw new AbortError() } - -function delay(ms: number, signal: AbortSignal) { - return new Promise((resolve, reject) => { - const timer = setTimeout(() => { - signal.removeEventListener('abort', onAbort) - resolve() - }, ms) - function onAbort() { - clearTimeout(timer) - reject(new AbortError()) - } - signal.addEventListener('abort', onAbort, {once: true}) - }) -} diff --git a/src/lib/media/video/multipart/uploadParts.test.ts b/src/lib/media/video/multipart/uploadParts.test.ts index f2553fd2a4..7965a28df9 100644 --- a/src/lib/media/video/multipart/uploadParts.test.ts +++ b/src/lib/media/video/multipart/uploadParts.test.ts @@ -1,4 +1,5 @@ import {AbortError} from '#/lib/async/cancelable' +import {MultipartUploadError} from './api' import {type ChunkReader, type UploadPartFn} from './types' import {uploadParts} from './uploadParts' @@ -66,7 +67,7 @@ describe('uploadParts', () => { const n = (attemptsByPart.get(part.partNumber) ?? 0) + 1 attemptsByPart.set(part.partNumber, n) if (part.partNumber === 2 && n === 1) { - return Promise.reject(new Error('transient')) + return Promise.reject(new TypeError('transient network error')) } return Promise.resolve({ partNumber: part.partNumber, @@ -87,9 +88,59 @@ describe('uploadParts', () => { expect(results).toHaveLength(3) }) + it('retries rate-limited parts', async () => { + let attempts = 0 + const uploadPart: UploadPartFn = ({part}) => { + attempts++ + if (attempts === 1) { + return Promise.reject( + new MultipartUploadError('rate limited', 'RateLimitExceeded', 429), + ) + } + return Promise.resolve({ + partNumber: part.partNumber, + sizeBytes: part.size, + }) + } + + await uploadParts({ + parts: parts.slice(0, 1), + reader: fakeReader(), + uploadPart, + totalBytes: 10, + setProgress: () => {}, + signal: new AbortController().signal, + }) + + expect(attempts).toBe(2) + }) + + it('does not retry a non-retryable response', async () => { + const uploadPart = jest.fn< + ReturnType, + Parameters + >(() => + Promise.reject( + new MultipartUploadError('bad request', 'InvalidRequest', 400), + ), + ) + + await expect( + uploadParts({ + parts: parts.slice(0, 1), + reader: fakeReader(), + uploadPart, + totalBytes: 10, + setProgress: () => {}, + signal: new AbortController().signal, + }), + ).rejects.toThrow('bad request') + expect(uploadPart).toHaveBeenCalledTimes(1) + }) + it('throws after exhausting attempts', async () => { const uploadPart: UploadPartFn = () => - Promise.reject(new Error('always fails')) + Promise.reject(new TypeError('always fails')) await expect( uploadParts({ diff --git a/src/lib/media/video/multipart/uploadParts.ts b/src/lib/media/video/multipart/uploadParts.ts index 6b42122158..67ab0ac7b4 100644 --- a/src/lib/media/video/multipart/uploadParts.ts +++ b/src/lib/media/video/multipart/uploadParts.ts @@ -1,5 +1,7 @@ import {AbortError} from '#/lib/async/cancelable' +import {isRetryableHttpStatus} from '#/lib/strings/errors' import {createProgressAggregator} from './aggregateProgress' +import {MultipartUploadError} from './api' import {MULTIPART_CONCURRENCY, MULTIPART_MAX_ATTEMPTS} from './constants' import { type ChunkReader, @@ -7,6 +9,7 @@ import { type PartUploadResult, type UploadPartFn, } from './types' +import {delay} from './utils' /** * Uploads every part with a concurrency cap and per-part retry, aggregating @@ -118,6 +121,7 @@ async function uploadPartWithRetry({ throw new AbortError() } lastError = err + if (!isRetryablePartError(err)) throw err if (attempt < maxAttempts) { await delay(500 * 2 ** (attempt - 1), signal) } @@ -126,16 +130,12 @@ async function uploadPartWithRetry({ throw lastError } -function delay(ms: number, signal: AbortSignal) { - return new Promise((resolve, reject) => { - const timer = setTimeout(() => { - signal.removeEventListener('abort', onAbort) - resolve() - }, ms) - function onAbort() { - clearTimeout(timer) - reject(new AbortError()) - } - signal.addEventListener('abort', onAbort, {once: true}) - }) +function isRetryablePartError(err: unknown) { + return ( + err instanceof TypeError || + (err instanceof MultipartUploadError && + (err.error === 'ServiceOverloaded' || + err.status === undefined || + isRetryableHttpStatus(err.status))) + ) } diff --git a/src/lib/media/video/multipart/utils.ts b/src/lib/media/video/multipart/utils.ts new file mode 100644 index 0000000000..e81e05642c --- /dev/null +++ b/src/lib/media/video/multipart/utils.ts @@ -0,0 +1,15 @@ +import {AbortError} from '#/lib/async/cancelable' + +export function delay(ms: number, signal: AbortSignal) { + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + signal.removeEventListener('abort', onAbort) + resolve() + }, ms) + function onAbort() { + clearTimeout(timer) + reject(new AbortError()) + } + signal.addEventListener('abort', onAbort, {once: true}) + }) +} diff --git a/src/lib/strings/errors.ts b/src/lib/strings/errors.ts index 994c91515b..11e7c11362 100644 --- a/src/lib/strings/errors.ts +++ b/src/lib/strings/errors.ts @@ -88,6 +88,10 @@ export function isCancelledError(e: unknown) { // TODO Replace this with error.shouldRetry() when available. -dsb const RETRYABLE_ERRORS = [408, 425, 429, 500, 502, 503, 504, 522, 524] -export function shouldRetryError(e: unknown) { - return e instanceof XRPCError && RETRYABLE_ERRORS.includes(e.status) +export function isRetryableHttpStatus(status: number) { + return RETRYABLE_ERRORS.includes(status) +} + +export function shouldRetryError(e: unknown) { + return e instanceof XRPCError && isRetryableHttpStatus(e.status) }