From a8fa7e8d17292b9f478fb9ca745dfe9b77a61e2b Mon Sep 17 00:00:00 2001 From: vineyardbovines Date: Tue, 28 Jul 2026 09:45:57 -0400 Subject: [PATCH] Address multipart upload review feedback --- src/lib/media/video/multipart/upload.ts | 52 ++++++++++++++----- .../media/video/multipart/uploadParts.test.ts | 27 ++++++++++ src/lib/media/video/multipart/uploadParts.ts | 7 ++- 3 files changed, 73 insertions(+), 13 deletions(-) diff --git a/src/lib/media/video/multipart/upload.ts b/src/lib/media/video/multipart/upload.ts index 1605bf40c9..fb6cc7a5c3 100644 --- a/src/lib/media/video/multipart/upload.ts +++ b/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 { 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 | undefined + // Kept outside the upload try block for finish-time missing-part recovery. let parts: ReturnType = [] 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 + getToken: (forceRefresh?: boolean) => Promise signal: AbortSignal resendMissingParts: (receivedPartNumbers: number[]) => Promise }): Promise { 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 | 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() } diff --git a/src/lib/media/video/multipart/uploadParts.test.ts b/src/lib/media/video/multipart/uploadParts.test.ts index edaa0bb536..f2553fd2a4 100644 --- a/src/lib/media/video/multipart/uploadParts.test.ts +++ b/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}) => { diff --git a/src/lib/media/video/multipart/uploadParts.ts b/src/lib/media/video/multipart/uploadParts.ts index ca7bcd2844..6b42122158 100644 --- a/src/lib/media/video/multipart/uploadParts.ts +++ b/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 }