Address multipart upload review feedback

src/lib/media/video/multipart/upload.ts

@@ -3,6 +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 {getServiceAuthToken} from '../upload.shared'
 import {mimeToExt} from '../util'
 import {
@@ -35,7 +36,7 @@ export async function uploadVideoMultipart({
   onStarted?: () => void
 }): Promise<AppBskyVideoDefs.JobStatus> {
   throwIfAborted(signal)
-  const tokenProvider = createTokenProvider(agent)
+  const tokenProvider = createTokenProvider(agent, signal)
   const token = await tokenProvider.get()
   const name = `${nanoid(12)}.${mimeToExt(video.mimeType)}`
   let session
@@ -60,6 +61,7 @@ export async function uploadVideoMultipart({
   }
   signal.addEventListener('abort', abortOnCancel, {once: true})
   let reader: ReturnType<typeof createChunkReader> | undefined
+  // Kept outside the upload try block for finish-time missing-part recovery.
   let parts: ReturnType<typeof planParts> = []
   try {
     try {
@@ -85,9 +87,6 @@ export async function uploadVideoMultipart({
       )
     }
 
-    // Finish stores this credential for the later PDS blob upload, so use a
-    // fresh token rather than the one that may have aged during transfer.
-    await tokenProvider.get(true)
     const activeReader = reader
     if (!activeReader) throw new Error('Video chunk reader is unavailable')
     return await finishAndRecover({
@@ -126,14 +125,18 @@ async function finishAndRecover({
   resendMissingParts,
 }: {
   jobId: string
-  getToken: () => Promise<string>
+  getToken: (forceRefresh?: boolean) => Promise<string>
   signal: AbortSignal
   resendMissingParts: (receivedPartNumbers: number[]) => Promise<boolean>
 }): Promise<AppBskyVideoDefs.JobStatus> {
   let createdFailures = 0
+  let forceTokenRefresh = true
   while (true) {
     throwIfAborted(signal)
-    const token = await getToken()
+    // Finish stores this credential for the later PDS blob upload. Refresh it
+    // once after part transfer, then reuse it while polling/recovering.
+    const token = await getToken(forceTokenRefresh)
+    forceTokenRefresh = false
     try {
       const result = await finishUpload(jobId, token, signal)
       return result.jobStatus
@@ -177,6 +180,11 @@ async function finishAndRecover({
             `Multipart upload ${status.state}`,
             status.state === 'aborted' ? 'UploadAborted' : 'UploadExpired',
           )
+        case 'completed':
+          throw new MultipartUploadError(
+            'Multipart upload completed without a job status',
+            'InvalidUploadStatus',
+          )
       }
     }
   }
@@ -233,7 +241,7 @@ async function abortThenFallbackOrResolve(
   )
 }
 
-function createTokenProvider(agent: AtpAgent) {
+function createTokenProvider(agent: AtpAgent, signal: AbortSignal) {
   let token: string | undefined
   let expiresAt = 0
   let refresh: Promise<string> | undefined
@@ -242,11 +250,7 @@ function createTokenProvider(agent: AtpAgent) {
     if (!forceRefresh && token && Date.now() < expiresAt - 60_000) return token
     if (!refresh) {
       const exp = Math.floor(Date.now() / 1000) + 60 * 30
-      refresh = getServiceAuthToken({
-        agent,
-        lxm: 'com.atproto.repo.uploadBlob',
-        exp,
-      })
+      refresh = getServiceAuthTokenWithRetry(agent, exp, signal)
         .then(nextToken => {
           token = nextToken
           expiresAt = exp * 1000
@@ -262,6 +266,30 @@ function createTokenProvider(agent: AtpAgent) {
   return {get}
 }
 
+async function getServiceAuthTokenWithRetry(
+  agent: AtpAgent,
+  exp: number,
+  signal: AbortSignal,
+) {
+  let lastError: unknown
+  for (let attempt = 1; attempt <= 3; attempt++) {
+    throwIfAborted(signal)
+    try {
+      return await getServiceAuthToken({
+        agent,
+        lxm: 'com.atproto.repo.uploadBlob',
+        exp,
+      })
+    } catch (err) {
+      throwIfAborted(signal)
+      if (!(err instanceof TypeError) && !shouldRetryError(err)) throw err
+      lastError = err
+      if (attempt < 3) await delay(500 * 2 ** (attempt - 1), signal)
+    }
+  }
+  throw lastError
+}
+
 function throwIfAborted(signal: AbortSignal) {
   if (signal.aborted) throw new AbortError()
 }

src/lib/media/video/multipart/uploadParts.test.ts

@@ -1,3 +1,4 @@
+import {AbortError} from '#/lib/async/cancelable'
 import {type ChunkReader, type UploadPartFn} from './types'
 import {uploadParts} from './uploadParts'
 
@@ -103,6 +104,32 @@ describe('uploadParts', () => {
     ).rejects.toThrow('always fails')
   })
 
+  it('preserves the originating error when sibling workers abort', async () => {
+    const uploadPart: UploadPartFn = ({part, signal}) => {
+      if (part.partNumber === 1) {
+        return new Promise((_resolve, reject) => {
+          signal.addEventListener('abort', () => reject(new AbortError()), {
+            once: true,
+          })
+        })
+      }
+      return Promise.reject(new Error('part upload failed'))
+    }
+
+    await expect(
+      uploadParts({
+        parts: parts.slice(0, 2),
+        reader: fakeReader(),
+        uploadPart,
+        totalBytes: 20,
+        setProgress: () => {},
+        signal: new AbortController().signal,
+        concurrency: 2,
+        maxAttempts: 1,
+      }),
+    ).rejects.toThrow('part upload failed')
+  })
+
   it('reports progress that reaches 1 when all parts complete', async () => {
     const progress: number[] = []
     const uploadPart: UploadPartFn = ({part, chunk, onProgress}) => {

src/lib/media/video/multipart/uploadParts.ts

@@ -78,10 +78,15 @@ export async function uploadParts({
     }),
   )
   signal.removeEventListener('abort', abortWorkers)
-  const failure = settled.find(
+  const failures = settled.filter(
     (result): result is PromiseRejectedResult => result.status === 'rejected',
   )
   if (signal.aborted) throw new AbortError()
+  // A sibling worker aborted after the first failure can settle earlier in
+  // array order. Preserve the originating error for fallback and telemetry.
+  const failure =
+    failures.find(result => !(result.reason instanceof AbortError)) ??
+    failures[0]
   if (failure) throw failure.reason
   return results
 }