From a35c4ac4bafea19e9b1c9330043f593d7867967b Mon Sep 17 00:00:00 2001 From: Spence Pope Date: Wed, 29 Jul 2026 12:54:03 -0400 Subject: [PATCH] APP-2670: add multipart video upload transport (#11222) --- src/analytics/features/types.ts | 1 + src/analytics/metrics/types.ts | 7 +- .../video/multipart/aggregateProgress.test.ts | 37 +++ .../video/multipart/aggregateProgress.ts | 19 ++ src/lib/media/video/multipart/api.ts | 134 ++++++++ src/lib/media/video/multipart/constants.ts | 13 + .../media/video/multipart/planParts.test.ts | 47 +++ src/lib/media/video/multipart/planParts.ts | 30 ++ src/lib/media/video/multipart/readChunk.ts | 21 ++ .../media/video/multipart/readChunk.web.ts | 27 ++ src/lib/media/video/multipart/types.ts | 72 +++++ src/lib/media/video/multipart/upload.ts | 292 ++++++++++++++++++ src/lib/media/video/multipart/uploadPart.ts | 94 ++++++ .../media/video/multipart/uploadParts.test.ts | 206 ++++++++++++ src/lib/media/video/multipart/uploadParts.ts | 129 ++++++++ src/lib/media/video/multipart/utils.ts | 27 ++ src/lib/media/video/telemetry.ts | 10 + src/lib/media/video/types.ts | 2 + src/lib/media/video/upload.ts | 27 +- src/lib/media/video/upload.web.ts | 27 +- src/lib/strings/errors.ts | 8 +- src/view/com/composer/state/video.ts | 16 +- 22 files changed, 1240 insertions(+), 6 deletions(-) create mode 100644 src/lib/media/video/multipart/aggregateProgress.test.ts create mode 100644 src/lib/media/video/multipart/aggregateProgress.ts create mode 100644 src/lib/media/video/multipart/api.ts create mode 100644 src/lib/media/video/multipart/constants.ts create mode 100644 src/lib/media/video/multipart/planParts.test.ts create mode 100644 src/lib/media/video/multipart/planParts.ts create mode 100644 src/lib/media/video/multipart/readChunk.ts create mode 100644 src/lib/media/video/multipart/readChunk.web.ts create mode 100644 src/lib/media/video/multipart/types.ts create mode 100644 src/lib/media/video/multipart/upload.ts create mode 100644 src/lib/media/video/multipart/uploadPart.ts create mode 100644 src/lib/media/video/multipart/uploadParts.test.ts create mode 100644 src/lib/media/video/multipart/uploadParts.ts create mode 100644 src/lib/media/video/multipart/utils.ts diff --git a/src/analytics/features/types.ts b/src/analytics/features/types.ts index 16cbe3a105..1d6cf131c1 100644 --- a/src/analytics/features/types.ts +++ b/src/analytics/features/types.ts @@ -19,6 +19,7 @@ export enum Features { PostThreadKnownLikersEnable = 'post_thread:known_likers:enable', PostThreadKnownLikersFetchEnable = 'post_thread:known_likers:fetch:enable', CustomLogoJapanEnable = 'custom_logo:japan:enable', + VideoMultipartUploadEnable = 'video:multipart_upload:enable', SearchStarterPacksV2Enable = 'search_starter_packs_v2:enable', FollowSortEnable = 'follow_sort:enable', diff --git a/src/analytics/metrics/types.ts b/src/analytics/metrics/types.ts index bd268603a0..12e0440846 100644 --- a/src/analytics/metrics/types.ts +++ b/src/analytics/metrics/types.ts @@ -5,7 +5,10 @@ import {type Platform} from 'react-native' import {type NotificationReason} from '#/lib/hooks/useNotificationHandler' -import {type VideoCompressSkipReason} from '#/lib/media/video/types' +import { + type VideoCompressSkipReason, + type VideoUploadTransport, +} from '#/lib/media/video/types' import {type NotificationType} from '#/state/queries/notifications/types' import {type FeedDescriptor} from '#/state/queries/post-feed' import {type LiveEventFeedMetricContext} from '#/features/liveEvents/types' @@ -1451,6 +1454,7 @@ export type Events = { bytes: number elapsedMs: number throughputBytesPerSec: number + transport: VideoUploadTransport } 'video:upload:uploadFailed': { uploadId: string @@ -1458,6 +1462,7 @@ export type Events = { bytes: number errorClass: string elapsedMs: number + transport: VideoUploadTransport } 'video:upload:processingStarted': { uploadId: string diff --git a/src/lib/media/video/multipart/aggregateProgress.test.ts b/src/lib/media/video/multipart/aggregateProgress.test.ts new file mode 100644 index 0000000000..a92273cbdb --- /dev/null +++ b/src/lib/media/video/multipart/aggregateProgress.test.ts @@ -0,0 +1,37 @@ +import {createProgressAggregator} from './aggregateProgress' + +describe('createProgressAggregator', () => { + it('sums bytes across parts against the total', () => { + const progress: number[] = [] + const report = createProgressAggregator(100, p => progress.push(p)) + + report(1, 50) + report(2, 25) + expect(progress).toEqual([0.5, 0.75]) + }) + + it('overwrites a part running count rather than adding it', () => { + const progress: number[] = [] + const report = createProgressAggregator(100, p => progress.push(p)) + + report(1, 20) + report(1, 40) + expect(progress).toEqual([0.2, 0.4]) + }) + + it('clamps to 1', () => { + const progress: number[] = [] + const report = createProgressAggregator(100, p => progress.push(p)) + + report(1, 150) + expect(progress).toEqual([1]) + }) + + it('reports 0 when the total is 0', () => { + const progress: number[] = [] + const report = createProgressAggregator(0, p => progress.push(p)) + + report(1, 10) + expect(progress).toEqual([0]) + }) +}) diff --git a/src/lib/media/video/multipart/aggregateProgress.ts b/src/lib/media/video/multipart/aggregateProgress.ts new file mode 100644 index 0000000000..c9e0b1914c --- /dev/null +++ b/src/lib/media/video/multipart/aggregateProgress.ts @@ -0,0 +1,19 @@ +/** + * Tracks bytes sent per part and reports overall progress (0..1) to the + * existing single-value progress callback. Parts upload concurrently, so each + * part reports its own running byte count and this sums them against the total. + */ +export function createProgressAggregator( + totalBytes: number, + setProgress: (progress: number) => void, +) { + const sentByPart = new Map() + return function reportPartProgress(partNumber: number, bytesSent: number) { + sentByPart.set(partNumber, bytesSent) + let sum = 0 + for (const value of sentByPart.values()) { + sum += value + } + setProgress(totalBytes > 0 ? Math.min(sum / totalBytes, 1) : 0) + } +} diff --git a/src/lib/media/video/multipart/api.ts b/src/lib/media/video/multipart/api.ts new file mode 100644 index 0000000000..811696b3e3 --- /dev/null +++ b/src/lib/media/video/multipart/api.ts @@ -0,0 +1,134 @@ +import {AbortError} from '#/lib/async/cancelable' +import {createVideoEndpointUrl} from '#/lib/media/video/util' +import { + type AbortUploadResponse, + type FinishUploadResponse, + type StartUploadResponse, + type UploadStatusResponse, +} from './types' + +export class MultipartUploadError extends Error { + constructor( + message: string, + public error?: string, + public status?: number, + ) { + super(message) + this.name = 'MultipartUploadError' + } +} + +async function request({ + route, + token, + signal, + method = 'POST', + body, + params, +}: { + route: string + token: string + signal?: AbortSignal + method?: 'GET' | 'POST' + body?: object + params?: Record +}): Promise { + if (signal?.aborted) throw new AbortError() + let res: Response + try { + res = await fetch(createVideoEndpointUrl(route, params), { + method, + headers: { + Authorization: `Bearer ${token}`, + ...(body ? {'Content-Type': 'application/json'} : {}), + }, + body: body ? JSON.stringify(body) : undefined, + signal, + }) + } catch (err) { + if (signal?.aborted) throw new AbortError() + throw err + } + const text = await res.text() + let data: unknown + try { + data = text ? JSON.parse(text) : undefined + } catch {} + if (!res.ok) { + const xrpc = data as {error?: string; message?: string} | undefined + throw new MultipartUploadError( + xrpc?.message || xrpc?.error || `Video service returned ${res.status}`, + xrpc?.error, + res.status, + ) + } + return data as T +} + +export function startUpload({ + token, + video, + name, + signal, +}: { + token: string + video: {size: number; mimeType: string} + name: string + signal: AbortSignal +}) { + return request({ + route: '/xrpc/app.bsky.video.startUpload', + token, + signal, + body: {sizeBytes: video.size, mimeType: video.mimeType, name}, + }) +} + +export function finishUpload( + jobId: string, + token: string, + signal: AbortSignal, +) { + return request({ + route: '/xrpc/app.bsky.video.finishUpload', + token, + signal, + body: {jobId}, + }) +} + +export function getUploadStatus( + jobId: string, + token: string, + signal?: AbortSignal, +) { + return request({ + route: '/xrpc/app.bsky.video.getUploadStatus', + token, + signal, + method: 'GET', + params: {jobId}, + }) +} + +export function abortUpload(jobId: string, token: string) { + return request({ + route: '/xrpc/app.bsky.video.abortUpload', + token, + body: {jobId}, + }) +} + +export function completedStatus(status: UploadStatusResponse) { + if ( + status.state !== 'completed' || + !status.completedJobId || + !status.jobStatus + ) { + return undefined + } + return { + completedJobId: status.completedJobId, + jobStatus: status.jobStatus, + } +} diff --git a/src/lib/media/video/multipart/constants.ts b/src/lib/media/video/multipart/constants.ts new file mode 100644 index 0000000000..2aaa37b8c2 --- /dev/null +++ b/src/lib/media/video/multipart/constants.ts @@ -0,0 +1,13 @@ +/* + * Multipart upload knobs. Tunable; part size must stay above the storage + * backend's minimum (R2/S3 require >= 5 MiB per part, except the last). + */ + +/** Max parts uploaded concurrently. */ +export const MULTIPART_CONCURRENCY = 3 + +/** Per-part upload attempts before the part (and the upload) fails. */ +export const MULTIPART_MAX_ATTEMPTS = 3 + +/** Attempts to begin/continue server-side finalization before checking state. */ +export const MULTIPART_FINISH_ATTEMPTS = 3 diff --git a/src/lib/media/video/multipart/planParts.test.ts b/src/lib/media/video/multipart/planParts.test.ts new file mode 100644 index 0000000000..fb7365fc74 --- /dev/null +++ b/src/lib/media/video/multipart/planParts.test.ts @@ -0,0 +1,47 @@ +import {getMissingParts, planParts} from './planParts' + +describe('planParts', () => { + it('splits an evenly divisible size into full parts', () => { + expect(planParts(20, 10)).toEqual([ + {partNumber: 1, offset: 0, size: 10}, + {partNumber: 2, offset: 10, size: 10}, + ]) + }) + + it('puts the remainder in the last part', () => { + expect(planParts(25, 10)).toEqual([ + {partNumber: 1, offset: 0, size: 10}, + {partNumber: 2, offset: 10, size: 10}, + {partNumber: 3, offset: 20, size: 5}, + ]) + }) + + it('returns a single part when the file is smaller than a part', () => { + expect(planParts(5, 10)).toEqual([{partNumber: 1, offset: 0, size: 5}]) + }) + + it('returns no parts for a non-positive size', () => { + expect(planParts(0, 10)).toEqual([]) + }) + + it('covers the whole file with no gaps or overlaps', () => { + const parts = planParts(1000, 128) + expect(parts[0].offset).toBe(0) + for (let i = 1; i < parts.length; i++) { + expect(parts[i].offset).toBe(parts[i - 1].offset + parts[i - 1].size) + } + const last = parts[parts.length - 1] + expect(last.offset + last.size).toBe(1000) + }) + + it('throws for a non-positive part size', () => { + expect(() => planParts(100, 0)).toThrow() + }) + + it('selects only parts the server has not received', () => { + const parts = planParts(25, 10) + expect(getMissingParts(parts, [1, 3])).toEqual([ + {partNumber: 2, offset: 10, size: 10}, + ]) + }) +}) diff --git a/src/lib/media/video/multipart/planParts.ts b/src/lib/media/video/multipart/planParts.ts new file mode 100644 index 0000000000..6dfaf29f1b --- /dev/null +++ b/src/lib/media/video/multipart/planParts.ts @@ -0,0 +1,30 @@ +import {type PartPlan} from './types' + +/** + * Splits a file of `totalSize` bytes into parts of at most `partSize` bytes. + * The last part carries the remainder. Part numbers are 1-indexed. Returns an + * empty list for a non-positive size. + */ +export function planParts(totalSize: number, partSize: number): PartPlan[] { + if (partSize <= 0) { + throw new Error('partSize must be positive') + } + const parts: PartPlan[] = [] + let offset = 0 + let partNumber = 1 + while (offset < totalSize) { + const size = Math.min(partSize, totalSize - offset) + parts.push({partNumber, offset, size}) + offset += size + partNumber += 1 + } + return parts +} + +export function getMissingParts( + parts: PartPlan[], + receivedPartNumbers: number[], +): PartPlan[] { + const received = new Set(receivedPartNumbers) + return parts.filter(part => !received.has(part.partNumber)) +} diff --git a/src/lib/media/video/multipart/readChunk.ts b/src/lib/media/video/multipart/readChunk.ts new file mode 100644 index 0000000000..9a8aaefe32 --- /dev/null +++ b/src/lib/media/video/multipart/readChunk.ts @@ -0,0 +1,21 @@ +import {File} from 'expo-file-system' + +import {type CompressedVideo} from '#/lib/media/video/types' +import {type ChunkReader} from './types' + +/** + * Native chunk reader. Opens one file handle and seeks per read, so the video + * bytes are never all held in JS memory. Call `close` when the upload finishes. + */ +export function createChunkReader(video: CompressedVideo): ChunkReader { + const handle = new File(video.uri).open() + return { + read(offset, size) { + handle.offset = offset + return Promise.resolve(handle.readBytes(size)) + }, + close() { + handle.close() + }, + } +} diff --git a/src/lib/media/video/multipart/readChunk.web.ts b/src/lib/media/video/multipart/readChunk.web.ts new file mode 100644 index 0000000000..8f3767f26c --- /dev/null +++ b/src/lib/media/video/multipart/readChunk.web.ts @@ -0,0 +1,27 @@ +import {type CompressedVideo} from '#/lib/media/video/types' +import {type ChunkReader} from './types' + +/** + * Web chunk reader. Web compression already produces the full buffer, so this + * slices it in memory. Falls back to fetching the uri once if `bytes` is + * missing. `close` is a no-op. + */ +export function createChunkReader(video: CompressedVideo): ChunkReader { + let bytesPromise: Promise | null = null + const getBytes = () => { + if (video.bytes) { + return Promise.resolve(video.bytes) + } + if (!bytesPromise) { + bytesPromise = fetch(video.uri).then(res => res.arrayBuffer()) + } + return bytesPromise + } + return { + async read(offset, size) { + const buffer = await getBytes() + return new Uint8Array(buffer, offset, size) + }, + close() {}, + } +} diff --git a/src/lib/media/video/multipart/types.ts b/src/lib/media/video/multipart/types.ts new file mode 100644 index 0000000000..16817778c3 --- /dev/null +++ b/src/lib/media/video/multipart/types.ts @@ -0,0 +1,72 @@ +/** + * One part of a multipart upload. `partNumber` is 1-indexed to match the S3 + * convention the backend uses. + */ +export type PartPlan = { + partNumber: number + offset: number + size: number +} + +/** Receipt returned by the video service after recording a part. */ +export type PartUploadResult = { + partNumber: number + sizeBytes: number +} + +/** + * Reads a byte range off the compressed video. Native opens a file handle and + * seeks; web slices an in-memory buffer. `close` releases the native handle and + * is a no-op on web. + */ +export type ChunkReader = { + read: (offset: number, size: number) => Promise + close: () => void +} + +/** + * Uploads one part through the video service's first-party proxy. + */ +export type UploadPartFn = (args: { + part: PartPlan + chunk: Uint8Array + onProgress: (bytesSent: number) => void + signal: AbortSignal +}) => Promise + +export type StartUploadResponse = { + jobId: string + partSizeBytes: number + partCount: number + expiresAt: string +} + +export type UploadState = + | 'created' + | 'finishing' + | 'completed' + | 'failed' + | 'aborted' + | 'expired' + +export type UploadStatusResponse = { + jobId: string + partSizeBytes: number + partCount: number + receivedParts: number[] + expiresAt: string + state: UploadState + completedJobId?: string + jobStatus?: import('@atproto/api').AppBskyVideoDefs.JobStatus + failureReason?: string +} + +export type FinishUploadResponse = { + completedJobId: string + jobStatus: import('@atproto/api').AppBskyVideoDefs.JobStatus +} + +export type AbortUploadResponse = Pick< + UploadStatusResponse, + 'completedJobId' | 'failureReason' +> & {state: 'aborted' | 'completed' | 'failed' | 'expired'} diff --git a/src/lib/media/video/multipart/upload.ts b/src/lib/media/video/multipart/upload.ts new file mode 100644 index 0000000000..af2d71d596 --- /dev/null +++ b/src/lib/media/video/multipart/upload.ts @@ -0,0 +1,292 @@ +import {type AppBskyVideoDefs, type AtpAgent} from '@atproto/api' +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 { + abortUpload, + completedStatus, + finishUpload, + getUploadStatus, + MultipartUploadError, + startUpload, +} from './api' +import {MULTIPART_FINISH_ATTEMPTS} from './constants' +import {getMissingParts, planParts} from './planParts' +import {createChunkReader} from './readChunk' +import {createUploadPart} from './uploadPart' +import {uploadParts} from './uploadParts' +import {delay, isRetryableMultipartError} from './utils' + +export class MultipartFallbackError extends Error {} + +export async function uploadVideoMultipart({ + video, + agent, + setProgress, + signal, + onStarted, +}: { + video: CompressedVideo + agent: AtpAgent + setProgress: (progress: number) => void + signal: AbortSignal + onStarted?: () => void +}): Promise { + throwIfAborted(signal) + const tokenProvider = createTokenProvider(agent, signal) + const token = await tokenProvider.get() + const name = `${nanoid(12)}.${mimeToExt(video.mimeType)}` + let session + try { + session = await startUpload({token, video, name, signal}) + } catch (err) { + if (signal.aborted) throw new AbortError() + // A server without multipart support, or one with the kill switch active, + // leaves no reservation behind. The legacy path remains authoritative. + throw new MultipartFallbackError( + err instanceof Error ? err.message : 'Multipart upload unavailable', + ) + } + onStarted?.() + + const {jobId} = session + const abortOnCancel = () => { + void tokenProvider + .get() + .then(currentToken => abortUpload(jobId, currentToken)) + .catch(() => {}) + } + 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 { + reader = createChunkReader(video) + parts = planParts(video.size, session.partSizeBytes) + if (parts.length !== session.partCount) { + throw new Error('Video service returned an invalid multipart plan') + } + await uploadParts({ + parts, + reader, + uploadPart: createUploadPart(jobId, tokenProvider.get), + totalBytes: video.size, + setProgress, + signal, + }) + } catch (err) { + if (signal.aborted) throw new AbortError() + return await abortThenFallbackOrResolve( + jobId, + await tokenProvider.get(), + err, + ) + } + + // Preserve TypeScript's narrowing inside the recovery callback. + const activeReader = reader + if (!activeReader) throw new Error('Video chunk reader is unavailable') + return await finishAndRecover({ + jobId, + getToken: tokenProvider.get, + signal, + resendMissingParts: async receivedPartNumbers => { + const missing = getMissingParts(parts, receivedPartNumbers) + if (missing.length === 0) return false + const missingBytes = missing.reduce((sum, part) => sum + part.size, 0) + const completedBytes = video.size - missingBytes + await uploadParts({ + parts: missing, + reader: activeReader, + uploadPart: createUploadPart(jobId, tokenProvider.get), + totalBytes: missingBytes, + setProgress: progress => + setProgress( + (completedBytes + progress * missingBytes) / video.size, + ), + signal, + }) + return true + }, + }) + } finally { + reader?.close() + signal.removeEventListener('abort', abortOnCancel) + } +} + +async function finishAndRecover({ + jobId, + getToken, + signal, + resendMissingParts, +}: { + jobId: string + getToken: (forceRefresh?: boolean) => Promise + signal: AbortSignal + resendMissingParts: (receivedPartNumbers: number[]) => Promise +}): Promise { + let createdFailures = 0 + let forceTokenRefresh = true + while (true) { + throwIfAborted(signal) + // 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 + } catch (finishError) { + throwIfAborted(signal) + const status = await getUploadStatusWithRetry(jobId, token, signal) + const completed = completedStatus(status) + if (completed) return completed.jobStatus + + switch (status.state) { + case 'created': + try { + const resentParts = await resendMissingParts(status.receivedParts) + if (resentParts) { + createdFailures = 0 + continue + } + } catch (err) { + throwIfAborted(signal) + return await abortThenFallbackOrResolve(jobId, token, err) + } + createdFailures++ + if (createdFailures < MULTIPART_FINISH_ATTEMPTS) { + await delay(500 * 2 ** (createdFailures - 1), signal) + continue + } + return await abortThenFallbackOrResolve(jobId, token, finishError) + case 'finishing': + // 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': + throw new MultipartUploadError( + status.failureReason || 'Multipart upload failed', + 'UploadFailed', + ) + case 'aborted': + case 'expired': + throw new MultipartUploadError( + `Multipart upload ${status.state}`, + status.state === 'aborted' ? 'UploadAborted' : 'UploadExpired', + ) + case 'completed': + throw new MultipartUploadError( + 'Multipart upload completed without a job status', + 'InvalidUploadStatus', + ) + default: + throw new MultipartUploadError( + 'Multipart upload returned an unknown status', + 'InvalidUploadStatus', + ) + } + } + } +} + +async function getUploadStatusWithRetry( + jobId: string, + token: string, + signal: AbortSignal, +) { + let lastError: unknown + for (let attempt = 1; attempt <= 3; attempt++) { + try { + return await getUploadStatus(jobId, token, signal) + } catch (err) { + throwIfAborted(signal) + if (!isRetryableMultipartError(err)) throw err + lastError = err + if (attempt < 3) await delay(500 * 2 ** (attempt - 1), signal) + } + } + throw lastError +} + +async function abortThenFallbackOrResolve( + jobId: string, + token: string, + cause: unknown, +): Promise { + const result = await abortUpload(jobId, token) + if (result.state === 'aborted') { + throw new MultipartFallbackError( + cause instanceof Error ? cause.message : 'Multipart upload failed', + ) + } + if (result.state === 'completed' && result.completedJobId) { + const status = await getUploadStatus(jobId, token) + const completed = completedStatus(status) + if (completed) return completed.jobStatus + } + throw new MultipartUploadError( + result.failureReason || `Multipart upload ${result.state}`, + result.state === 'failed' ? 'UploadFailed' : undefined, + ) +} + +function createTokenProvider(agent: AtpAgent, signal: AbortSignal) { + let token: string | undefined + let expiresAt = 0 + let refresh: Promise | undefined + + async function get(forceRefresh = false) { + if (!forceRefresh && token && Date.now() < expiresAt - 60_000) return token + if (!refresh) { + const exp = Math.floor(Date.now() / 1000) + 60 * 30 + refresh = getServiceAuthTokenWithRetry(agent, exp, signal) + .then(nextToken => { + token = nextToken + expiresAt = exp * 1000 + return nextToken + }) + .finally(() => { + refresh = undefined + }) + } + return refresh + } + + 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/uploadPart.ts b/src/lib/media/video/multipart/uploadPart.ts new file mode 100644 index 0000000000..e3c57fc2b2 --- /dev/null +++ b/src/lib/media/video/multipart/uploadPart.ts @@ -0,0 +1,94 @@ +import {AbortError} from '#/lib/async/cancelable' +import {createVideoEndpointUrl} from '#/lib/media/video/util' +import {MultipartUploadError} from './api' +import {type UploadPartFn} from './types' + +export function createUploadPart( + jobId: string, + getToken: (forceRefresh?: boolean) => Promise, +): UploadPartFn { + return async args => { + try { + return await sendPart(jobId, await getToken(), args) + } catch (err) { + if ( + err instanceof MultipartUploadError && + (err.status === 401 || err.error === 'AuthRequired') + ) { + args.onProgress(0) + return await sendPart(jobId, await getToken(true), args) + } + throw err + } + } +} + +function sendPart( + jobId: string, + token: string, + {part, chunk, onProgress, signal}: Parameters[0], +) { + return new Promise>>((resolve, reject) => { + if (signal.aborted) { + reject(new AbortError()) + return + } + const xhr = new XMLHttpRequest() + const abort = () => xhr.abort() + signal.addEventListener('abort', abort, {once: true}) + const cleanup = () => signal.removeEventListener('abort', abort) + + xhr.upload.addEventListener('progress', event => { + onProgress(event.loaded) + }) + xhr.onerror = () => { + cleanup() + reject(new TypeError('Network request failed')) + } + xhr.onabort = () => { + cleanup() + reject(new AbortError()) + } + xhr.onload = () => { + cleanup() + let data: { + partNumber?: number + sizeBytes?: number + error?: string + message?: string + } + try { + data = JSON.parse(xhr.responseText) + } catch { + data = {} + } + if (xhr.status < 200 || xhr.status >= 300) { + reject( + new MultipartUploadError( + data.message || + data.error || + `Video service returned ${xhr.status}`, + data.error, + xhr.status, + ), + ) + } else { + onProgress(part.size) + resolve({ + partNumber: data.partNumber ?? part.partNumber, + sizeBytes: data.sizeBytes ?? part.size, + }) + } + } + xhr.open( + 'POST', + createVideoEndpointUrl('/xrpc/app.bsky.video.uploadPart', { + jobId, + partNumber: String(part.partNumber), + }), + ) + xhr.setRequestHeader('Content-Type', 'application/octet-stream') + xhr.setRequestHeader('Authorization', `Bearer ${token}`) + xhr.send(chunk as XMLHttpRequestBodyInit) + }) +} diff --git a/src/lib/media/video/multipart/uploadParts.test.ts b/src/lib/media/video/multipart/uploadParts.test.ts new file mode 100644 index 0000000000..7965a28df9 --- /dev/null +++ b/src/lib/media/video/multipart/uploadParts.test.ts @@ -0,0 +1,206 @@ +import {AbortError} from '#/lib/async/cancelable' +import {MultipartUploadError} from './api' +import {type ChunkReader, type UploadPartFn} from './types' +import {uploadParts} from './uploadParts' + +function fakeReader(): ChunkReader { + return { + read: (_offset, size) => Promise.resolve(new Uint8Array(size)), + close: () => {}, + } +} + +const parts = [ + {partNumber: 1, offset: 0, size: 10}, + {partNumber: 2, offset: 10, size: 10}, + {partNumber: 3, offset: 20, size: 5}, +] + +describe('uploadParts', () => { + it('uploads every part and returns results ordered by part number', async () => { + const uploadPart: UploadPartFn = ({part}) => + Promise.resolve({ + partNumber: part.partNumber, + sizeBytes: part.size, + }) + + const results = await uploadParts({ + parts, + reader: fakeReader(), + uploadPart, + totalBytes: 25, + setProgress: () => {}, + signal: new AbortController().signal, + }) + + expect(results.map(r => r.partNumber)).toEqual([1, 2, 3]) + expect(results.map(r => r.sizeBytes)).toEqual([10, 10, 5]) + }) + + it('respects the concurrency cap', async () => { + let active = 0 + let maxActive = 0 + const uploadPart: UploadPartFn = async ({part}) => { + active++ + maxActive = Math.max(maxActive, active) + await new Promise(r => setTimeout(r, 5)) + active-- + return {partNumber: part.partNumber, sizeBytes: part.size} + } + + await uploadParts({ + parts, + reader: fakeReader(), + uploadPart, + totalBytes: 25, + setProgress: () => {}, + signal: new AbortController().signal, + concurrency: 2, + }) + + expect(maxActive).toBeLessThanOrEqual(2) + }) + + it('retries a failing part and succeeds', async () => { + const attemptsByPart = new Map() + const uploadPart: UploadPartFn = ({part}) => { + const n = (attemptsByPart.get(part.partNumber) ?? 0) + 1 + attemptsByPart.set(part.partNumber, n) + if (part.partNumber === 2 && n === 1) { + return Promise.reject(new TypeError('transient network error')) + } + return Promise.resolve({ + partNumber: part.partNumber, + sizeBytes: part.size, + }) + } + + const results = await uploadParts({ + parts, + reader: fakeReader(), + uploadPart, + totalBytes: 25, + setProgress: () => {}, + signal: new AbortController().signal, + }) + + expect(attemptsByPart.get(2)).toBe(2) + 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 TypeError('always fails')) + + await expect( + uploadParts({ + parts, + reader: fakeReader(), + uploadPart, + totalBytes: 25, + setProgress: () => {}, + signal: new AbortController().signal, + maxAttempts: 2, + }), + ).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}) => { + onProgress(chunk.byteLength) + return Promise.resolve({ + partNumber: part.partNumber, + sizeBytes: part.size, + }) + } + + await uploadParts({ + parts, + reader: fakeReader(), + uploadPart, + totalBytes: 25, + setProgress: p => progress.push(p), + signal: new AbortController().signal, + concurrency: 1, + }) + + expect(progress[progress.length - 1]).toBe(1) + }) +}) diff --git a/src/lib/media/video/multipart/uploadParts.ts b/src/lib/media/video/multipart/uploadParts.ts new file mode 100644 index 0000000000..cda502fb54 --- /dev/null +++ b/src/lib/media/video/multipart/uploadParts.ts @@ -0,0 +1,129 @@ +import {AbortError} from '#/lib/async/cancelable' +import {createProgressAggregator} from './aggregateProgress' +import {MULTIPART_CONCURRENCY, MULTIPART_MAX_ATTEMPTS} from './constants' +import { + type ChunkReader, + type PartPlan, + type PartUploadResult, + type UploadPartFn, +} from './types' +import {delay, isRetryableMultipartError} from './utils' + +/** + * Uploads every part with a concurrency cap and per-part retry, aggregating + * byte progress into `setProgress`. Reads each chunk lazily just before its + * upload so only `concurrency` chunks are in memory at once. Resolves with the + * part results ordered by part number. + */ +export async function uploadParts({ + parts, + reader, + uploadPart, + totalBytes, + setProgress, + signal, + concurrency = MULTIPART_CONCURRENCY, + maxAttempts = MULTIPART_MAX_ATTEMPTS, +}: { + parts: PartPlan[] + reader: ChunkReader + uploadPart: UploadPartFn + totalBytes: number + setProgress: (progress: number) => void + signal: AbortSignal + concurrency?: number + maxAttempts?: number +}): Promise { + const reportPartProgress = createProgressAggregator(totalBytes, setProgress) + const results: PartUploadResult[] = new Array(parts.length) + const workerController = new AbortController() + const abortWorkers = () => workerController.abort() + signal.addEventListener('abort', abortWorkers, {once: true}) + const workerSignal = workerController.signal + + let nextIndex = 0 + async function worker() { + while (true) { + if (workerSignal.aborted) { + throw new AbortError() + } + const index = nextIndex++ + if (index >= parts.length) { + return + } + const part = parts[index] + const chunk = await reader.read(part.offset, part.size) + results[index] = await uploadPartWithRetry({ + part, + chunk, + uploadPart, + maxAttempts, + signal: workerSignal, + onProgress: bytesSent => reportPartProgress(part.partNumber, bytesSent), + }) + } + } + + const workers = Array.from( + {length: Math.min(concurrency, parts.length)}, + () => worker(), + ) + const settled = await Promise.allSettled( + workers.map(async workerPromise => { + try { + await workerPromise + } catch (err) { + workerController.abort() + throw err + } + }), + ) + signal.removeEventListener('abort', abortWorkers) + 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 +} + +async function uploadPartWithRetry({ + part, + chunk, + uploadPart, + maxAttempts, + signal, + onProgress, +}: { + part: PartPlan + chunk: Uint8Array + uploadPart: UploadPartFn + maxAttempts: number + signal: AbortSignal + onProgress: (bytesSent: number) => void +}): Promise { + let lastError: unknown + for (let attempt = 1; attempt <= maxAttempts; attempt++) { + if (signal.aborted) { + throw new AbortError() + } + try { + return await uploadPart({part, chunk, onProgress, signal}) + } catch (err) { + if (signal.aborted) { + throw new AbortError() + } + lastError = err + if (!isRetryableMultipartError(err)) throw err + if (attempt < maxAttempts) { + await delay(500 * 2 ** (attempt - 1), signal) + } + } + } + throw lastError +} diff --git a/src/lib/media/video/multipart/utils.ts b/src/lib/media/video/multipart/utils.ts new file mode 100644 index 0000000000..4184ab5666 --- /dev/null +++ b/src/lib/media/video/multipart/utils.ts @@ -0,0 +1,27 @@ +import {AbortError} from '#/lib/async/cancelable' +import {isRetryableHttpStatus} from '#/lib/strings/errors' +import {MultipartUploadError} from './api' + +export function isRetryableMultipartError(err: unknown) { + return ( + err instanceof TypeError || + (err instanceof MultipartUploadError && + (err.error === 'ServiceOverloaded' || + err.status === undefined || + isRetryableHttpStatus(err.status))) + ) +} + +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/media/video/telemetry.ts b/src/lib/media/video/telemetry.ts index 67ff77c58e..9181ec002b 100644 --- a/src/lib/media/video/telemetry.ts +++ b/src/lib/media/video/telemetry.ts @@ -5,6 +5,7 @@ import {nanoid} from 'nanoid/non-secure' import { type ProbedMetadata, type VideoCompressSkipReason, + type VideoUploadTransport, } from '#/lib/media/video/types' import {Sentry} from '#/logger/sentry/lib' import {type Metrics} from '#/analytics/metrics' @@ -45,6 +46,7 @@ export type VideoTelemetry = { compressCompleted: (video: {size: number; mimeType: string}) => void compressFailed: (e: unknown) => void uploadStarted: (bytes: number) => void + uploadTransport: (transport: VideoUploadTransport) => void uploadCompleted: (jobId: string) => void uploadFailed: (e: unknown) => void processingStarted: (jobId: string) => void @@ -70,6 +72,7 @@ export function createVideoTelemetry({ let phaseStartedAt = startedAt let jobId: string | undefined let uploadBytes: number | undefined + let uploadTransport: VideoUploadTransport = 'legacy' let txnEnded = false let abortBound = true @@ -226,6 +229,11 @@ export function createVideoTelemetry({ metric('video:upload:uploadStarted', {uploadId, engine, bytes}) }, + uploadTransport(transport) { + uploadTransport = transport + phaseSpan?.setAttribute('video.upload.transport', transport) + }, + uploadCompleted(id) { jobId = id const elapsedMs = Date.now() - phaseStartedAt @@ -238,6 +246,7 @@ export function createVideoTelemetry({ elapsedMs, throughputBytesPerSec: elapsedMs > 0 ? Math.round((bytes * 1000) / elapsedMs) : 0, + transport: uploadTransport, }) endPhaseSpan() phase = undefined @@ -250,6 +259,7 @@ export function createVideoTelemetry({ bytes: uploadBytes ?? 0, errorClass: errorClass(e), elapsedMs: Date.now() - phaseStartedAt, + transport: uploadTransport, }) endTxn('error') detachAbort() diff --git a/src/lib/media/video/types.ts b/src/lib/media/video/types.ts index 6825f03ba3..1d2062ce00 100644 --- a/src/lib/media/video/types.ts +++ b/src/lib/media/video/types.ts @@ -8,6 +8,8 @@ export type VideoCompressSkipReason = | 'no-webcodecs' | 'compress-error-fallback' +export type VideoUploadTransport = 'multipart' | 'legacy' | 'legacy-fallback' + export type CompressedVideo = { uri: string mimeType: string diff --git a/src/lib/media/video/upload.ts b/src/lib/media/video/upload.ts index 721ee7f94f..b91ad7a153 100644 --- a/src/lib/media/video/upload.ts +++ b/src/lib/media/video/upload.ts @@ -6,7 +6,12 @@ import {nanoid} from 'nanoid/non-secure' import {AbortError} from '#/lib/async/cancelable' import {ServerError} from '#/lib/media/video/errors' -import {type CompressedVideo} from '#/lib/media/video/types' +import { + type CompressedVideo, + type VideoUploadTransport, +} from '#/lib/media/video/types' +import {Features, features} from '#/analytics/features' +import {MultipartFallbackError, uploadVideoMultipart} from './multipart/upload' import {getServiceAuthToken, getVideoUploadLimits} from './upload.shared' import {createVideoEndpointUrl, mimeToExt} from './util' @@ -17,6 +22,7 @@ export async function uploadVideo({ setProgress, signal, i18n, + onTransport, }: { video: CompressedVideo agent: AtpAgent @@ -24,12 +30,31 @@ export async function uploadVideo({ setProgress: (progress: number) => void signal: AbortSignal i18n: I18n + onTransport?: (transport: VideoUploadTransport) => void }) { if (signal.aborted) { throw new AbortError() } await getVideoUploadLimits(agent, i18n) + if (features.isOn(Features.VideoMultipartUploadEnable)) { + try { + return await uploadVideoMultipart({ + video, + agent, + setProgress, + signal, + onStarted: () => onTransport?.('multipart'), + }) + } catch (err) { + if (!(err instanceof MultipartFallbackError)) throw err + onTransport?.('legacy-fallback') + setProgress(0) + } + } else { + onTransport?.('legacy') + } + const uri = createVideoEndpointUrl('/xrpc/app.bsky.video.uploadVideo', { did, name: `${nanoid(12)}.${mimeToExt(video.mimeType)}`, diff --git a/src/lib/media/video/upload.web.ts b/src/lib/media/video/upload.web.ts index e88a04707c..cfefbd797d 100644 --- a/src/lib/media/video/upload.web.ts +++ b/src/lib/media/video/upload.web.ts @@ -5,7 +5,12 @@ import {nanoid} from 'nanoid/non-secure' import {AbortError} from '#/lib/async/cancelable' import {ServerError} from '#/lib/media/video/errors' -import {type CompressedVideo} from '#/lib/media/video/types' +import { + type CompressedVideo, + type VideoUploadTransport, +} from '#/lib/media/video/types' +import {Features, features} from '#/analytics/features' +import {MultipartFallbackError, uploadVideoMultipart} from './multipart/upload' import {getServiceAuthToken, getVideoUploadLimits} from './upload.shared' import {createVideoEndpointUrl, mimeToExt} from './util' @@ -16,6 +21,7 @@ export async function uploadVideo({ setProgress, signal, i18n, + onTransport, }: { video: CompressedVideo agent: AtpAgent @@ -23,12 +29,31 @@ export async function uploadVideo({ setProgress: (progress: number) => void signal: AbortSignal i18n: I18n + onTransport?: (transport: VideoUploadTransport) => void }) { if (signal.aborted) { throw new AbortError() } await getVideoUploadLimits(agent, i18n) + if (features.isOn(Features.VideoMultipartUploadEnable)) { + try { + return await uploadVideoMultipart({ + video, + agent, + setProgress, + signal, + onStarted: () => onTransport?.('multipart'), + }) + } catch (err) { + if (!(err instanceof MultipartFallbackError)) throw err + onTransport?.('legacy-fallback') + setProgress(0) + } + } else { + onTransport?.('legacy') + } + const uri = createVideoEndpointUrl('/xrpc/app.bsky.video.uploadVideo', { did, name: `${nanoid(12)}.${mimeToExt(video.mimeType)}`, 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) } diff --git a/src/view/com/composer/state/video.ts b/src/view/com/composer/state/video.ts index 492eac1c5d..ed4384ccda 100644 --- a/src/view/com/composer/state/video.ts +++ b/src/view/com/composer/state/video.ts @@ -326,6 +326,7 @@ export async function processVideo( did, signal, i18n, + onTransport: telemetry.uploadTransport, setProgress: p => { dispatch({type: 'update_progress', progress: p, signal}) }, @@ -387,7 +388,7 @@ export async function processVideo( telemetry.processingFailed(e) dispatch({ type: 'to_error', - error: i18n._(msg`Video failed to process`), + error: getProcessingErrorMessage(status?.error, i18n), signal, }) return // Exit async loop @@ -420,6 +421,19 @@ export async function processVideo( } } +function getProcessingErrorMessage(error: string | undefined, i18n: I18n) { + switch (error) { + case 'video_too_long': + return i18n._(msg`The selected video is too long.`) + case 'bad_aspect_ratio': + return i18n._(msg`The selected video has an unsupported aspect ratio.`) + case 'unsupported_codec': + return i18n._(msg`The selected video uses an unsupported format.`) + default: + return i18n._(msg`Video failed to process`) + } +} + function getCompressErrorMessage(e: unknown, i18n: I18n): string | null { if (e instanceof AbortError) { return null