fix: recover failed and legacy Loom imports

apps/media-server/Dockerfile

@@ -16,6 +16,7 @@ RUN bun test src/__tests__/lib/audio-quality-policy.test.ts src/__tests__/lib/au
 	&& bun test src/__tests__/lib/audio-quality-formats.integration.test.ts src/__tests__/lib/audio-quality-benchmark.test.ts \
 	&& bun test src/__tests__/lib/audio-levels.integration.test.ts \
 	&& bun test src/__tests__/lib/drive-resumable-upload.test.ts src/__tests__/lib/storage-upload.test.ts src/__tests__/lib/container-memory.test.ts src/__tests__/lib/parallel-download.test.ts \
+	&& bun test src/__tests__/lib/loom-import-limit.test.ts \
 	&& bun test src/__tests__/lib/media-size.test.ts src/__tests__/lib/media-probe.integration.test.ts \
 	&& bun test src/__tests__/lib/media-audio.integration.test.ts \
 	&& bun test src/__tests__/lib/recording-verification.integration.test.ts src/__tests__/lib/job-manager.test.ts \

apps/media-server/src/__tests__/lib/loom-import-limit.test.ts

@@ -0,0 +1,170 @@
+import { afterEach, describe, expect, test } from "bun:test";
+import { execFileSync } from "node:child_process";
+import { mkdtemp, rm } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import {
+	downloadVideoToTemp,
+	LOOM_IMPORT_MAX_BYTES,
+} from "../../lib/media-video";
+
+const originalFetch = globalThis.fetch;
+const originalDownloadConcurrency =
+	process.env.MEDIA_SERVER_DOWNLOAD_CONCURRENCY;
+
+function stubFetch(response: () => Promise<Response>) {
+	globalThis.fetch = Object.assign(response, {
+		preconnect: originalFetch.preconnect,
+	});
+}
+
+describe("Loom import file size", () => {
+	afterEach(() => {
+		globalThis.fetch = originalFetch;
+		if (originalDownloadConcurrency === undefined) {
+			delete process.env.MEDIA_SERVER_DOWNLOAD_CONCURRENCY;
+		} else {
+			process.env.MEDIA_SERVER_DOWNLOAD_CONCURRENCY =
+				originalDownloadConcurrency;
+		}
+	});
+
+	test("permits exactly the configured boundary", async () => {
+		expect(LOOM_IMPORT_MAX_BYTES).toBe(10 * 1024 ** 3);
+		stubFetch(async () => new Response(new Uint8Array(16)));
+		const result = await downloadVideoToTemp(
+			"https://source.test/video.mp4",
+			".mp4",
+			undefined,
+			16,
+		);
+		try {
+			expect(Bun.file(result.path).size).toBe(16);
+		} finally {
+			await result.cleanup();
+		}
+	});
+
+	test("rejects an oversized declared body before reading it", async () => {
+		let cancelled = false;
+		stubFetch(
+			async () =>
+				new Response(
+					new ReadableStream({
+						cancel() {
+							cancelled = true;
+						},
+					}),
+					{ headers: { "Content-Length": "17" } },
+				),
+		);
+		await expect(
+			downloadVideoToTemp(
+				"https://source.test/video.mp4",
+				".mp4",
+				undefined,
+				16,
+			),
+		).rejects.toThrow("10 GB");
+		expect(cancelled).toBe(true);
+	});
+
+	test("enforces the import limit when parallel downloads are enabled", async () => {
+		process.env.MEDIA_SERVER_DOWNLOAD_CONCURRENCY = "4";
+		stubFetch(async () => new Response(new Uint8Array(17)));
+		await expect(
+			downloadVideoToTemp(
+				"https://source.test/video.mp4",
+				".mp4",
+				undefined,
+				16,
+			),
+		).rejects.toThrow("10 GB");
+	});
+
+	test("cancels a chunked transfer when it exceeds the limit", async () => {
+		let cancelled = false;
+		stubFetch(
+			async () =>
+				new Response(
+					new ReadableStream({
+						start(controller) {
+							controller.enqueue(new Uint8Array(8));
+							controller.enqueue(new Uint8Array(9));
+						},
+						cancel() {
+							cancelled = true;
+						},
+					}),
+				),
+		);
+		await expect(
+			downloadVideoToTemp(
+				"https://source.test/video.mp4",
+				".mp4",
+				undefined,
+				16,
+			),
+		).rejects.toThrow("10 GB");
+		expect(cancelled).toBe(true);
+	});
+
+	test("rejects an oversized real HLS recording without returning truncated media", async () => {
+		const directory = await mkdtemp(join(tmpdir(), "cap-loom-limit-"));
+		let server: ReturnType<typeof Bun.serve> | undefined;
+		try {
+			execFileSync("ffmpeg", [
+				"-hide_banner",
+				"-loglevel",
+				"error",
+				"-y",
+				"-f",
+				"lavfi",
+				"-i",
+				"testsrc2=size=64x64:rate=10",
+				"-t",
+				"2",
+				"-c:v",
+				"libx264",
+				"-g",
+				"10",
+				"-f",
+				"hls",
+				"-hls_time",
+				"1",
+				"-hls_segment_filename",
+				join(directory, "segment%d.ts"),
+				join(directory, "index.m3u8"),
+			]);
+			globalThis.fetch = originalFetch;
+			server = Bun.serve({
+				hostname: "127.0.0.1",
+				port: 0,
+				fetch(request) {
+					const path = new URL(request.url).pathname;
+					return /^\/(?:index\.m3u8|segment\d+\.ts)$/.test(path)
+						? new Response(Bun.file(join(directory, path.slice(1))))
+						: new Response(null, { status: 404 });
+				},
+			});
+			const source = `${server.url}index.m3u8`;
+			const complete = await downloadVideoToTemp(
+				source,
+				".m3u8",
+				undefined,
+				1024 ** 2,
+			);
+			try {
+				expect(Bun.file(complete.path).size).toBeGreaterThan(1);
+			} finally {
+				await complete.cleanup();
+			}
+			await expect(
+				downloadVideoToTemp(source, ".m3u8", undefined, 1),
+			).rejects.toThrow("10 GB");
+		} finally {
+			server?.stop(true);
+			await rm(directory, { recursive: true, force: true });
+		}
+	});
+});

apps/media-server/src/__tests__/lib/media-routes-real-world.integration.test.ts

@@ -1237,6 +1237,81 @@ describe("media routes real-world integration tests", () => {
 		30_000,
 	);
 
+	test("recovers a saved output after preview failure without overwriting it", async () => {
+		const thumbnail = spyOn(mediaVideo, "generateThumbnail").mockRejectedValue(
+			new Error("Thumbnail decode interrupted"),
+		);
+		const jobs: string[] = [];
+		try {
+			const body = {
+				videoId: "saved-loom-output",
+				userId: "real-process-user",
+				videoUrl: fixtureUrl(),
+				outputPresignedUrl: uploadUrl("saved-output.mp4"),
+				thumbnailPresignedUrl: uploadUrl("saved-output.jpg"),
+				inputExtension: ".mp4",
+				maxImportBytes: 10 * 1024 ** 3,
+			};
+			const first = await app.fetch(mediaPostRequest("/video/process", body));
+			const firstJobId = ((await first.json()) as { jobId: string }).jobId;
+			jobs.push(firstJobId);
+			expect((await waitForTerminalJob(firstJobId)).phase).toBe("error");
+			const original = Uint8Array.from(
+				uploadedBytes("/uploads/saved-output.mp4"),
+			);
+			const second = await app.fetch(
+				mediaPostRequest("/video/process", {
+					...body,
+					existingOutput: {
+						url: uploadUrl("saved-output.mp4"),
+						objectIdentity: objectIdentity(original),
+						fileSize: original.byteLength,
+					},
+				}),
+			);
+			expect(second.status).toBe(200);
+			const secondJobId = ((await second.json()) as { jobId: string }).jobId;
+			jobs.push(secondJobId);
+			expect((await waitForTerminalJob(secondJobId)).phase).toBe("complete");
+			expect(uploadRequests.filter((path) => path.endsWith(".mp4"))).toEqual([
+				"/uploads/saved-output.mp4",
+			]);
+			expect(uploadedBytes("/uploads/saved-output.mp4")).toEqual(original);
+		} finally {
+			thumbnail.mockRestore();
+			for (const jobId of jobs) deleteJob(jobId);
+		}
+	}, 30_000);
+
+	test("holds a saved output whose storage identity changed without overwriting it", async () => {
+		const original = new Uint8Array(
+			await Bun.file(TEST_VIDEO_WITH_AUDIO).arrayBuffer(),
+		);
+		uploadedArtifacts.set("/uploads/changed-output.mp4", original);
+		const response = await app.fetch(
+			mediaPostRequest("/video/process", {
+				videoId: "changed-output",
+				userId: "real-process-user",
+				videoUrl: fixtureUrl(),
+				outputPresignedUrl: uploadUrl("changed-output.mp4"),
+				maxImportBytes: 10 * 1024 ** 3,
+				existingOutput: {
+					url: uploadUrl("changed-output.mp4"),
+					objectIdentity: '"previous-identity"',
+					fileSize: original.byteLength,
+				},
+			}),
+		);
+		const { jobId } = (await response.json()) as { jobId: string };
+		try {
+			expect((await waitForTerminalJob(jobId)).phase).toBe("error");
+			expect(uploadRequests).toHaveLength(0);
+			expect(uploadedBytes("/uploads/changed-output.mp4")).toEqual(original);
+		} finally {
+			deleteJob(jobId);
+		}
+	}, 30_000);
+
 	test("still fails processing when the video output cannot be uploaded", async () => {
 		uploadFailures.set("/uploads/failed-output.mp4", 403);
 		const response = await app.fetch(

apps/media-server/src/__tests__/lib/storage-upload.test.ts

@@ -1,6 +1,6 @@
 import { afterEach, describe, expect, test } from "bun:test";
 import { createHash } from "node:crypto";
-import { mkdtemp, rm, writeFile } from "node:fs/promises";
+import { mkdtemp, open, rm, writeFile } from "node:fs/promises";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
 import {
@@ -32,6 +32,71 @@ describe("uploadFileToStorage", () => {
 		globalThis.fetch = originalFetch;
 	});
 
+	test("uploads a 10 GB sparse file as 160 bounded parts with its Loom attempt", async () => {
+		const uploadFile = await createTempUploadFile(1);
+		const handle = await open(uploadFile.path, "r+");
+		try {
+			await handle.truncate(10 * 1024 ** 3);
+		} finally {
+			await handle.close();
+		}
+		let uploadedParts = 0;
+		let uploadedBytes = 0;
+		let completed = false;
+		globalThis.fetch = (async (input, init) => {
+			const url = String(input);
+			if (url.endsWith("/sign-part")) {
+				const request = JSON.parse(String(init?.body));
+				expect(request.loomImportAttemptId).toBe("loom-run:0");
+				expect(request.contentLength).toBe(64 * 1024 ** 2);
+				return Response.json({
+					url: `https://storage.test/part/${request.partNumber}`,
+				});
+			}
+			if (init?.method === "PUT") {
+				const body = init.body;
+				if (!(body instanceof Blob))
+					throw new Error("Expected a bounded file slice");
+				uploadedParts++;
+				uploadedBytes += body.size;
+				expect(body.size).toBe(64 * 1024 ** 2);
+				return new Response(null, {
+					headers: { ETag: `"part-${uploadedParts}"` },
+				});
+			}
+			if (url.endsWith("/complete")) {
+				const request = JSON.parse(String(init?.body));
+				expect(request.loomImportAttemptId).toBe("loom-run:0");
+				expect(request.parts).toHaveLength(160);
+				completed = true;
+				return Response.json({ objectIdentity: '"multipart-result"' });
+			}
+			throw new Error("Unexpected multipart request");
+		}) as typeof fetch;
+		try {
+			await uploadFileToStorage(
+				uploadFile.path,
+				{
+					type: "multipart",
+					videoId: "video",
+					key: "owner/video/raw-upload.mp4",
+					uploadId: "upload",
+					partSize: 64 * 1024 ** 2,
+					signPartUrl: "https://cap.test/sign-part",
+					completeUrl: "https://cap.test/complete",
+					abortUrl: "https://cap.test/abort",
+					loomImportAttemptId: "loom-run:0",
+				},
+				"video/mp4",
+			);
+			expect(uploadedParts).toBe(160);
+			expect(uploadedBytes).toBe(10 * 1024 ** 3);
+			expect(completed).toBe(true);
+		} finally {
+			await uploadFile.cleanup();
+		}
+	});
+
 	test.each([false, true])(
 		"checks full remote bytes after Drive metadata version drift (corrupt=%s)",
 		async (corrupt) => {

apps/media-server/src/__tests__/routes/video.test.ts

@@ -51,6 +51,8 @@ describe("GET /video/status", () => {
 		expect(data).toHaveProperty("canAcceptNewProbeProcess");
 		expect(data).toHaveProperty("jobCount");
 		expect(data).toHaveProperty("jobs");
+		expect(data.loomMultipartUploads).toBe(true);
+		expect(data.loomImportMaxBytes).toBe(10 * 1024 ** 3);
 		expect(Array.isArray(data.jobs)).toBe(true);
 	});
 });
@@ -404,6 +406,44 @@ describe("POST /video/process", () => {
 		mock.restore();
 	});
 
+	test("accepts a Loom multipart original without a single-PUT original URL", async () => {
+		spyOn(jobManager, "canAcceptNewVideoProcess").mockReturnValue(false);
+		const response = await app.fetch(
+			videoPostRequest("/video/import", {
+				videoId: "test-id",
+				userId: "user-id",
+				videoUrl: "https://example.com/video.mp4",
+				outputPresignedUrl: "https://s3.example.com/output",
+				maxImportBytes: 10 * 1024 ** 3,
+				sourceUpload: {
+					type: "multipart",
+					videoId: "test-id",
+					loomImportAttemptId: "run:0",
+					key: "user-id/test-id/raw-upload.mp4",
+					uploadId: "upload-id",
+					partSize: 64 * 1024 ** 2,
+					signPartUrl: "https://cap.example.com/sign",
+					completeUrl: "https://cap.example.com/complete",
+					abortUrl: "https://cap.example.com/abort",
+				},
+			}),
+		);
+		expect(response.status).toBe(503);
+		expect((await response.json()).code).toBe("SERVER_BUSY");
+	});
+
+	test("rejects an import without an original upload target", async () => {
+		const response = await app.fetch(
+			videoPostRequest("/video/import", {
+				videoId: "test-id",
+				userId: "user-id",
+				videoUrl: "https://example.com/video.mp4",
+				outputPresignedUrl: "https://s3.example.com/output",
+			}),
+		);
+		expect(response.status).toBe(400);
+	});
+
 	test("returns 400 for missing required fields", async () => {
 		const response = await app.fetch(videoPostRequest("/video/process", {}));