Address remaining multipart review feedback

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 service may have assembled the upload even though the finish
-          // the object. Retrying is idempotent; legacy fallback is unsafe.
+          // 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<void>((resolve, reject) => {
-    const timer = setTimeout(() => {
-      signal.removeEventListener('abort', onAbort)
-      resolve()
-    }, ms)
-    function onAbort() {
-      clearTimeout(timer)
-      reject(new AbortError())
-    }
-    signal.addEventListener('abort', onAbort, {once: true})
-  })
-}

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<UploadPartFn>,
+      Parameters<UploadPartFn>
+    >(() =>
+      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({

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) {
+function isRetryablePartError(err: unknown) {
-  return new Promise<void>((resolve, reject) => {
+  return (
-    const timer = setTimeout(() => {
+    err instanceof TypeError ||
-      signal.removeEventListener('abort', onAbort)
+    (err instanceof MultipartUploadError &&
-      resolve()
+      (err.error === 'ServiceOverloaded' ||
-    }, ms)
+        err.status === undefined ||
-    function onAbort() {
+        isRetryableHttpStatus(err.status)))
-      clearTimeout(timer)
+  )
-      reject(new AbortError())
-    }
-    signal.addEventListener('abort', onAbort, {once: true})
-  })
 }

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<void>((resolve, reject) => {
+    const timer = setTimeout(() => {
+      signal.removeEventListener('abort', onAbort)
+      resolve()
+    }, ms)
+    function onAbort() {
+      clearTimeout(timer)
+      reject(new AbortError())
+    }
+    signal.addEventListener('abort', onAbort, {once: true})
+  })
+}

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) {
+export function isRetryableHttpStatus(status: number) {
-  return e instanceof XRPCError && RETRYABLE_ERRORS.includes(e.status)
+  return RETRYABLE_ERRORS.includes(status)
+}
+
+export function shouldRetryError(e: unknown) {
+  return e instanceof XRPCError && isRetryableHttpStatus(e.status)
 }