APP-2670: add multipart video upload transport

This commit is contained in:
vineyardbovines
2026-07-21 11:20:14 -04:00
parent b586127ea7
commit e807cc5b54
13 changed files with 577 additions and 33 deletions
+1
View File
@@ -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',
AATest = 'aa-test',
}
+134
View File
@@ -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<T>({
route,
token,
signal,
method = 'POST',
body,
params,
}: {
route: string
token: string
signal?: AbortSignal
method?: 'GET' | 'POST'
body?: object
params?: Record<string, string>
}): Promise<T> {
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<StartUploadResponse>({
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<FinishUploadResponse>({
route: '/xrpc/app.bsky.video.finishUpload',
token,
signal,
body: {jobId},
})
}
export function getUploadStatus(
jobId: string,
token: string,
signal?: AbortSignal,
) {
return request<UploadStatusResponse>({
route: '/xrpc/app.bsky.video.getUploadStatus',
token,
signal,
method: 'GET',
params: {jobId},
})
}
export function abortUpload(jobId: string, token: string) {
return request<AbortUploadResponse>({
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,
}
}
+3 -9
View File
@@ -3,17 +3,11 @@
* backend's minimum (R2/S3 require >= 5 MiB per part, except the last).
*/
/** Target size per part. */
export const MULTIPART_PART_SIZE = 8 * 1024 * 1024 // 8 MiB
/**
* Files below this skip multipart and use the single-shot POST. A file that
* would be a single part gains nothing from the multipart machinery.
*/
export const MULTIPART_MIN_FILE_SIZE = 16 * 1024 * 1024 // 16 MiB
/** 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
@@ -1,4 +1,4 @@
import {planParts} from './planParts'
import {getMissingParts, planParts} from './planParts'
describe('planParts', () => {
it('splits an evenly divisible size into full parts', () => {
@@ -37,4 +37,11 @@ describe('planParts', () => {
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},
])
})
})
@@ -20,3 +20,11 @@ export function planParts(totalSize: number, partSize: number): PartPlan[] {
}
return parts
}
export function getMissingParts(
parts: PartPlan[],
receivedPartNumbers: number[],
): PartPlan[] {
const received = new Set(receivedPartNumbers)
return parts.filter(part => !received.has(part.partNumber))
}
+40 -10
View File
@@ -8,14 +8,10 @@ export type PartPlan = {
size: number
}
/**
* Result of uploading a single part. `etag` is the value the storage backend
* (R2) returns for the part; the complete request lists these back to assemble
* the object.
*/
/** Receipt returned by the video service after recording a part. */
export type PartUploadResult = {
partNumber: number
etag: string
sizeBytes: number
}
/**
@@ -29,10 +25,7 @@ export type ChunkReader = {
}
/**
* Uploads one part to storage and resolves with its ETag. This is the only
* protocol-specific seam - for the presigned R2 approach it does a direct PUT
* to the presigned URL and reads the ETag off the response. Injected into the
* orchestrator so the transport can be filled in once the backend lands.
* Uploads one part through the video service's first-party proxy.
*/
export type UploadPartFn = (args: {
part: PartPlan
@@ -40,3 +33,40 @@ export type UploadPartFn = (args: {
onProgress: (bytesSent: number) => void
signal: AbortSignal
}) => Promise<PartUploadResult>
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'}
+248
View File
@@ -0,0 +1,248 @@
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 {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'
export class MultipartFallbackError extends Error {}
export async function uploadVideoMultipart({
video,
agent,
setProgress,
signal,
}: {
video: CompressedVideo
agent: AtpAgent
setProgress: (progress: number) => void
signal: AbortSignal
}): Promise<AppBskyVideoDefs.JobStatus> {
throwIfAborted(signal)
let token = await mintToken(agent)
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',
)
}
const {jobId} = session
const abortOnCancel = () => {
void abortUpload(jobId, token).catch(() => {})
}
signal.addEventListener('abort', abortOnCancel, {once: true})
let reader: ReturnType<typeof createChunkReader> | undefined
let parts: ReturnType<typeof planParts> = []
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, token),
totalBytes: video.size,
setProgress,
signal,
})
} catch (err) {
if (signal.aborted) throw new AbortError()
return await abortThenFallbackOrResolve(jobId, token, err)
}
// 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.
token = await mintToken(agent)
const activeReader = reader
if (!activeReader) throw new Error('Video chunk reader is unavailable')
return await finishAndRecover({
jobId,
token,
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, token),
totalBytes: missingBytes,
setProgress: progress =>
setProgress(
(completedBytes + progress * missingBytes) / video.size,
),
signal,
})
return true
},
})
} finally {
reader?.close()
signal.removeEventListener('abort', abortOnCancel)
}
}
async function finishAndRecover({
jobId,
token,
signal,
resendMissingParts,
}: {
jobId: string
token: string
signal: AbortSignal
resendMissingParts: (receivedPartNumbers: number[]) => Promise<boolean>
}): Promise<AppBskyVideoDefs.JobStatus> {
let createdFailures = 0
while (true) {
throwIfAborted(signal)
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':
// Finalization owns the reservation and may already have assembled
// the object. Retrying is idempotent; legacy fallback is unsafe.
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',
)
}
}
}
}
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 (!isRetryableStatusError(err)) throw err
lastError = err
if (attempt < 3) await delay(500 * 2 ** (attempt - 1), signal)
}
}
throw lastError
}
function isRetryableStatusError(err: unknown) {
return (
err instanceof TypeError ||
(err instanceof MultipartUploadError &&
(err.error === 'ServiceOverloaded' ||
err.status === undefined ||
err.status >= 500))
)
}
async function abortThenFallbackOrResolve(
jobId: string,
token: string,
cause: unknown,
): Promise<AppBskyVideoDefs.JobStatus> {
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 mintToken(agent: AtpAgent) {
return getServiceAuthToken({
agent,
lxm: 'com.atproto.repo.uploadBlob',
exp: Date.now() / 1000 + 60 * 30,
})
}
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})
})
}
@@ -0,0 +1,71 @@
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, token: string): UploadPartFn {
return ({part, chunk, onProgress, signal}) =>
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)
})
}
@@ -19,7 +19,7 @@ describe('uploadParts', () => {
const uploadPart: UploadPartFn = ({part}) =>
Promise.resolve({
partNumber: part.partNumber,
etag: `etag-${part.partNumber}`,
sizeBytes: part.size,
})
const results = await uploadParts({
@@ -32,7 +32,7 @@ describe('uploadParts', () => {
})
expect(results.map(r => r.partNumber)).toEqual([1, 2, 3])
expect(results.map(r => r.etag)).toEqual(['etag-1', 'etag-2', 'etag-3'])
expect(results.map(r => r.sizeBytes)).toEqual([10, 10, 5])
})
it('respects the concurrency cap', async () => {
@@ -43,7 +43,7 @@ describe('uploadParts', () => {
maxActive = Math.max(maxActive, active)
await new Promise(r => setTimeout(r, 5))
active--
return {partNumber: part.partNumber, etag: `e${part.partNumber}`}
return {partNumber: part.partNumber, sizeBytes: part.size}
}
await uploadParts({
@@ -69,7 +69,7 @@ describe('uploadParts', () => {
}
return Promise.resolve({
partNumber: part.partNumber,
etag: `e${part.partNumber}`,
sizeBytes: part.size,
})
}
@@ -109,7 +109,7 @@ describe('uploadParts', () => {
onProgress(chunk.byteLength)
return Promise.resolve({
partNumber: part.partNumber,
etag: `e${part.partNumber}`,
sizeBytes: part.size,
})
}
+23 -7
View File
@@ -12,10 +12,7 @@ import {
* 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, ready for the complete request.
*
* `uploadPart` is the transport seam - it stays stubbed until the presigned R2
* PUT is wired up.
* part results ordered by part number.
*/
export async function uploadParts({
parts,
@@ -38,11 +35,15 @@ export async function uploadParts({
}): Promise<PartUploadResult[]> {
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 (signal.aborted) {
if (workerSignal.aborted) {
throw new AbortError()
}
const index = nextIndex++
@@ -56,7 +57,7 @@ export async function uploadParts({
chunk,
uploadPart,
maxAttempts,
signal,
signal: workerSignal,
onProgress: bytesSent => reportPartProgress(part.partNumber, bytesSent),
})
}
@@ -66,7 +67,22 @@ export async function uploadParts({
{length: Math.min(concurrency, parts.length)},
() => worker(),
)
await Promise.all(workers)
const settled = await Promise.allSettled(
workers.map(async workerPromise => {
try {
await workerPromise
} catch (err) {
workerController.abort()
throw err
}
}),
)
signal.removeEventListener('abort', abortWorkers)
const failure = settled.find(
(result): result is PromiseRejectedResult => result.status === 'rejected',
)
if (signal.aborted) throw new AbortError()
if (failure) throw failure.reason
return results
}
+11
View File
@@ -7,6 +7,8 @@ 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 {Features, features} from '#/analytics/features'
import {MultipartFallbackError, uploadVideoMultipart} from './multipart/upload'
import {getServiceAuthToken, getVideoUploadLimits} from './upload.shared'
import {createVideoEndpointUrl, mimeToExt} from './util'
@@ -30,6 +32,15 @@ export async function uploadVideo({
}
await getVideoUploadLimits(agent, i18n)
if (features.isOn(Features.VideoMultipartUploadEnable)) {
try {
return await uploadVideoMultipart({video, agent, setProgress, signal})
} catch (err) {
if (!(err instanceof MultipartFallbackError)) throw err
setProgress(0)
}
}
const uri = createVideoEndpointUrl('/xrpc/app.bsky.video.uploadVideo', {
did,
name: `${nanoid(12)}.${mimeToExt(video.mimeType)}`,
+11
View File
@@ -6,6 +6,8 @@ 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 {Features, features} from '#/analytics/features'
import {MultipartFallbackError, uploadVideoMultipart} from './multipart/upload'
import {getServiceAuthToken, getVideoUploadLimits} from './upload.shared'
import {createVideoEndpointUrl, mimeToExt} from './util'
@@ -29,6 +31,15 @@ export async function uploadVideo({
}
await getVideoUploadLimits(agent, i18n)
if (features.isOn(Features.VideoMultipartUploadEnable)) {
try {
return await uploadVideoMultipart({video, agent, setProgress, signal})
} catch (err) {
if (!(err instanceof MultipartFallbackError)) throw err
setProgress(0)
}
}
const uri = createVideoEndpointUrl('/xrpc/app.bsky.video.uploadVideo', {
did,
name: `${nanoid(12)}.${mimeToExt(video.mimeType)}`,
+14 -1
View File
@@ -387,7 +387,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 +420,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