APP-2670: add multipart video upload transport (#11222)

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',
 

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

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])
+  })
+})

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<number, number>()
+  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)
+  }
+}

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<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,
+  }
+}

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

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},
+    ])
+  })
+})

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))
+}

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()
+    },
+  }
+}

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<ArrayBuffer> | 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() {},
+  }
+}

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<Uint8Array>
+  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<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'}

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<AppBskyVideoDefs.JobStatus> {
+  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<typeof createChunkReader> | undefined
+  // Kept outside the upload try block for finish-time missing-part recovery.
+  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, 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<string>
+  signal: AbortSignal
+  resendMissingParts: (receivedPartNumbers: number[]) => Promise<boolean>
+}): Promise<AppBskyVideoDefs.JobStatus> {
+  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<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 createTokenProvider(agent: AtpAgent, signal: AbortSignal) {
+  let token: string | undefined
+  let expiresAt = 0
+  let refresh: Promise<string> | 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()
+}

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<string>,
+): 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<UploadPartFn>[0],
+) {
+  return new Promise<Awaited<ReturnType<UploadPartFn>>>((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)
+  })
+}

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<number, number>()
+    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<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 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)
+  })
+})

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<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 (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<PartUploadResult> {
+  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
+}

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<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/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()

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

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)}`,

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)}`,

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)
 }

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