From b586127ea75792f5e63b7a4ab710dde297b58e1a Mon Sep 17 00:00:00 2001 From: vineyardbovines Date: Fri, 10 Jul 2026 10:31:32 -0400 Subject: [PATCH] scaffold multipart video upload primitives --- .../video/multipart/aggregateProgress.test.ts | 37 +++++ .../video/multipart/aggregateProgress.ts | 19 +++ src/lib/media/video/multipart/constants.ts | 19 +++ .../media/video/multipart/planParts.test.ts | 40 ++++++ src/lib/media/video/multipart/planParts.ts | 22 +++ src/lib/media/video/multipart/readChunk.ts | 21 +++ .../media/video/multipart/readChunk.web.ts | 27 ++++ src/lib/media/video/multipart/types.ts | 42 ++++++ .../media/video/multipart/uploadParts.test.ts | 128 ++++++++++++++++++ src/lib/media/video/multipart/uploadParts.ts | 120 ++++++++++++++++ 10 files changed, 475 insertions(+) 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/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/uploadParts.test.ts create mode 100644 src/lib/media/video/multipart/uploadParts.ts 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/constants.ts b/src/lib/media/video/multipart/constants.ts new file mode 100644 index 0000000000..e89aa885b9 --- /dev/null +++ b/src/lib/media/video/multipart/constants.ts @@ -0,0 +1,19 @@ +/* + * Multipart upload knobs. Tunable; part size must stay above the storage + * 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 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..683aa8ebf9 --- /dev/null +++ b/src/lib/media/video/multipart/planParts.test.ts @@ -0,0 +1,40 @@ +import {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() + }) +}) diff --git a/src/lib/media/video/multipart/planParts.ts b/src/lib/media/video/multipart/planParts.ts new file mode 100644 index 0000000000..633c300dab --- /dev/null +++ b/src/lib/media/video/multipart/planParts.ts @@ -0,0 +1,22 @@ +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 +} 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..1b0415b008 --- /dev/null +++ b/src/lib/media/video/multipart/types.ts @@ -0,0 +1,42 @@ +/** + * 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 +} + +/** + * 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. + */ +export type PartUploadResult = { + partNumber: number + etag: string +} + +/** + * 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 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. + */ +export type UploadPartFn = (args: { + part: PartPlan + chunk: Uint8Array + onProgress: (bytesSent: number) => void + signal: AbortSignal +}) => Promise 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..f27622a585 --- /dev/null +++ b/src/lib/media/video/multipart/uploadParts.test.ts @@ -0,0 +1,128 @@ +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, + etag: `etag-${part.partNumber}`, + }) + + 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.etag)).toEqual(['etag-1', 'etag-2', 'etag-3']) + }) + + 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, etag: `e${part.partNumber}`} + } + + 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 Error('transient')) + } + return Promise.resolve({ + partNumber: part.partNumber, + etag: `e${part.partNumber}`, + }) + } + + 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('throws after exhausting attempts', async () => { + const uploadPart: UploadPartFn = () => + Promise.reject(new Error('always fails')) + + await expect( + uploadParts({ + parts, + reader: fakeReader(), + uploadPart, + totalBytes: 25, + setProgress: () => {}, + signal: new AbortController().signal, + maxAttempts: 2, + }), + ).rejects.toThrow('always fails') + }) + + 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, + etag: `e${part.partNumber}`, + }) + } + + 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..8bfba91a7b --- /dev/null +++ b/src/lib/media/video/multipart/uploadParts.ts @@ -0,0 +1,120 @@ +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' + +/** + * 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. + */ +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) + + let nextIndex = 0 + async function worker() { + while (true) { + if (signal.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, + onProgress: bytesSent => reportPartProgress(part.partNumber, bytesSent), + }) + } + } + + const workers = Array.from( + {length: Math.min(concurrency, parts.length)}, + () => worker(), + ) + await Promise.all(workers) + 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 (attempt < maxAttempts) { + await delay(500 * 2 ** (attempt - 1), signal) + } + } + } + throw lastError +} + +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}) + }) +}