fix: preserve playback through media processing failures

Author: richiemcilroy

.github/workflows/recording-reliability.yml

@@ -44,4 +44,7 @@ jobs:
             __tests__/unit/desktop-recording-*.test.ts \
             __tests__/unit/desktop-segments-*.test.ts \
             __tests__/unit/finalize-desktop-recording.test.ts \
-            __tests__/unit/media-server-progress.test.ts
+            __tests__/unit/media-server-progress.test.ts \
+            __tests__/unit/media-processing-budget.test.ts \
+            __tests__/unit/playback-source.test.ts \
+            __tests__/unit/upload-progress-playback.test.ts

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

@@ -19,6 +19,7 @@ import type appType from "../../app";
 import * as containerCpu from "../../lib/container-cpu";
 import * as containerMemory from "../../lib/container-memory";
 import type { Job, JobProgress } from "../../lib/job-manager";
+import { withTimeout } from "../../lib/media-common";
 import { probeVideoFile } from "../../lib/media-probe";
 import * as mediaVideo from "../../lib/media-video";
 import * as recordingVerification from "../../lib/recording-verification";
@@ -40,6 +41,9 @@ let baseUrl = "";
 let tempDir = "";
 
 const uploadedArtifacts = new Map<string, Uint8Array>();
+const uploadFailures = new Map<string, number>();
+const uploadRequests: string[] = [];
+const fixtureReads: string[] = [];
 const recordingSources = new Map<string, Uint8Array>();
 const sourceReads: {
 	path: string;
@@ -364,6 +368,7 @@ beforeAll(async () => {
 								: null;
 
 				if (fixturePath) {
+					if (request.method === "GET") fixtureReads.push(url.pathname);
 					const fixture = Bun.file(fixturePath);
 					const headers = {
 						"Content-Type": "video/mp4",
@@ -408,6 +413,11 @@ beforeAll(async () => {
 			}
 
 			if (request.method === "PUT" && url.pathname.startsWith("/uploads/")) {
+				uploadRequests.push(url.pathname);
+				const failureStatus = uploadFailures.get(url.pathname);
+				if (failureStatus) {
+					return new Response("Storage unavailable", { status: failureStatus });
+				}
 				uploadConditions.push(request.headers.get("if-none-match"));
 				if (
 					request.headers.get("if-none-match") === "*" &&
@@ -451,6 +461,9 @@ beforeEach(() => {
 		pressure: 0.0625,
 	});
 	uploadedArtifacts.clear();
+	uploadFailures.clear();
+	uploadRequests.length = 0;
+	fixtureReads.length = 0;
 	transientFixtureFailures = 0;
 	permanentFixtureFailures = 0;
 	slowFixtureCancellations = 0;
@@ -950,6 +963,128 @@ describe("media routes real-world integration tests", () => {
 		}
 	}, 90000);
 
+	test.each([403, 503])(
+		"keeps a playable processed video when thumbnail storage returns %s",
+		async (status) => {
+			uploadFailures.set("/uploads/optional-thumbnail.jpg", status);
+			const response = await app.fetch(
+				mediaPostRequest("/video/process", {
+					videoId: "optional-thumbnail",
+					userId: "real-process-user",
+					videoUrl: fixtureUrl(),
+					outputPresignedUrl: uploadUrl("optional-thumbnail.mp4"),
+					thumbnailPresignedUrl: uploadUrl("optional-thumbnail.jpg"),
+					inputExtension: ".mp4",
+				}),
+			);
+			expect(response.status).toBe(200);
+			const { jobId } = (await response.json()) as { jobId: string };
+			try {
+				const job = await waitForTerminalJob(jobId);
+				expect(job.phase).toBe("complete");
+				expect(job.error).toBeUndefined();
+				expect(fixtureReads).toEqual(["/fixtures/test-with-audio.mp4"]);
+				expect(uploadRequests.filter((path) => path.endsWith(".mp4"))).toEqual([
+					"/uploads/optional-thumbnail.mp4",
+				]);
+				expect(
+					uploadRequests.filter((path) => path.endsWith(".jpg")),
+				).toHaveLength(status === 403 ? 1 : 5);
+				const output = join(tempDir, `optional-thumbnail-${status}.mp4`);
+				await writeFile(
+					output,
+					uploadedBytes("/uploads/optional-thumbnail.mp4"),
+				);
+				execFileSync("ffmpeg", [
+					"-v",
+					"error",
+					"-xerror",
+					"-i",
+					output,
+					"-f",
+					"null",
+					"-",
+				]);
+				const metadata = await probeVideoFile(output);
+				expect(metadata.videoCodec).toBe("h264");
+				expect(metadata.audioCodec).toBe("aac");
+			} 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(
+			mediaPostRequest("/video/process", {
+				videoId: "failed-output",
+				userId: "real-process-user",
+				videoUrl: fixtureUrl(),
+				outputPresignedUrl: uploadUrl("failed-output.mp4"),
+				thumbnailPresignedUrl: uploadUrl("failed-output.jpg"),
+				inputExtension: ".mp4",
+			}),
+		);
+		expect(response.status).toBe(200);
+		const { jobId } = (await response.json()) as { jobId: string };
+		try {
+			const job = await waitForTerminalJob(jobId);
+			expect(job.phase).toBe("error");
+			expect(job.error).toContain("403");
+			expect(uploadRequests).toEqual(["/uploads/failed-output.mp4"]);
+			expect(uploadedArtifacts.size).toBe(0);
+		} finally {
+			deleteJob(jobId);
+		}
+	}, 30_000);
+
+	test("does not complete a cancelled job during thumbnail upload", async () => {
+		let started: (() => void) | undefined;
+		const ready = new Promise<void>((resolve) => {
+			started = resolve;
+		});
+		const upload = spyOn(mediaVideo, "uploadToS3").mockImplementation(
+			async (_data, _url, _contentType, signal) => {
+				if (!signal) throw new Error("Missing thumbnail upload signal");
+				await new Promise<void>((_resolve, reject) => {
+					signal.addEventListener("abort", () => reject(signal.reason), {
+						once: true,
+					});
+					started?.();
+				});
+			},
+		);
+		let jobId: string | undefined;
+		try {
+			const response = await app.fetch(
+				mediaPostRequest("/video/process", {
+					videoId: "cancelled-thumbnail",
+					userId: "real-process-user",
+					videoUrl: fixtureUrl(),
+					outputPresignedUrl: uploadUrl("cancelled-thumbnail.mp4"),
+					thumbnailPresignedUrl: uploadUrl("cancelled-thumbnail.jpg"),
+					inputExtension: ".mp4",
+				}),
+			);
+			expect(response.status).toBe(200);
+			jobId = ((await response.json()) as { jobId: string }).jobId;
+			await withTimeout(ready, 10_000);
+			getJob(jobId)?.abortController?.abort(new Error("Worker cancelled"));
+			const job = await waitForTerminalJob(jobId);
+			expect(job.phase).toBe("error");
+			expect(job.error).toBe("Worker cancelled");
+			expect(upload).toHaveBeenCalledTimes(1);
+		} finally {
+			if (jobId) {
+				getJob(jobId)?.abortController?.abort();
+				deleteJob(jobId);
+			}
+			upload.mockRestore();
+		}
+	}, 15_000);
+
 	test("retries transient segment downloads and completes a real mux job", async () => {
 		const response = await app.fetch(
 			mediaPostRequest("/video/mux-segments", {

apps/media-server/src/__tests__/lib/media-video.integration.test.ts

@@ -302,6 +302,33 @@ describe("recording upload cancellation", () => {
 });
 
 describe("generateThumbnail integration tests", () => {
+	test("uses the first frame when a sparse video has no frame after the thumbnail seek", async () => {
+		const directory = mkdtempSync(join(tmpdir(), "cap-sparse-thumbnail-"));
+		const input = join(directory, "single-frame.mp4");
+		try {
+			execFileSync("ffmpeg", [
+				"-v",
+				"error",
+				"-f",
+				"lavfi",
+				"-i",
+				"color=c=blue:s=160x120:r=1/2",
+				"-frames:v",
+				"1",
+				"-c:v",
+				"libx264",
+				"-pix_fmt",
+				"yuv420p",
+				input,
+			]);
+			const thumbnail = await generateThumbnail(input, 2);
+			expect(thumbnail.length).toBeGreaterThan(0);
+			expect([...thumbnail.subarray(0, 2)]).toEqual([0xff, 0xd8]);
+		} finally {
+			rmSync(directory, { recursive: true, force: true });
+		}
+	});
+
 	test("joins an in-flight thumbnail decoder when its worker is cancelled", async () => {
 		let ready: (() => void) | undefined;
 		const started = new Promise<void>((resolve) => {

apps/media-server/src/lib/media-video.ts

@@ -1,12 +1,13 @@
 import { createHash, randomUUID } from "node:crypto";
 import { mkdtemp, readdir, rm, writeFile } from "node:fs/promises";
-import { join } from "node:path";
+import { isAbsolute, join } from "node:path";
 import { setTimeout as sleep } from "node:timers/promises";
 import { type BunFile, file, spawn } from "bun";
 import { uploadDriveResumable } from "./drive-resumable-upload";
 import type { VideoMetadata } from "./job-manager";
 import {
 	DOWNLOAD_TIMEOUT_MS,
+	normalizeLocalPath,
 	PROCESS_TIMEOUT_MS,
 	type ProgressCallback,
 	UPLOAD_TIMEOUT_MS,
@@ -1670,6 +1671,8 @@ function getThumbnailTimestamp(
 	return Math.min(Math.max(0, timestamp), Math.max(0, duration - 0.1));
 }
 
+class EmptyThumbnailError extends Error {}
+
 export async function generateThumbnail(
 	inputPath: string,
 	duration: number,
@@ -1679,12 +1682,53 @@ export async function generateThumbnail(
 	abortSignal?.throwIfAborted();
 	const opts = { ...DEFAULT_THUMBNAIL_OPTIONS, ...options };
 	const timestamp = getThumbnailTimestamp(duration, opts.timestamp);
+	const deadline = performance.now() + THUMBNAIL_TIMEOUT_MS;
+	try {
+		return await generateThumbnailFrame(
+			inputPath,
+			opts,
+			timestamp,
+			THUMBNAIL_TIMEOUT_MS,
+			abortSignal,
+		);
+	} catch (error) {
+		abortSignal?.throwIfAborted();
+		const remainingMs = deadline - performance.now();
+		if (
+			!(error instanceof EmptyThumbnailError) ||
+			!isAbsolute(normalizeLocalPath(inputPath)) ||
+			timestamp <= 0 ||
+			remainingMs <= 0
+		) {
+			throw error;
+		}
+		return await generateThumbnailFrame(
+			inputPath,
+			opts,
+			0,
+			remainingMs,
+			abortSignal,
+			true,
+		);
+	}
+}
+
+async function generateThumbnailFrame(
+	inputPath: string,
+	opts: Required<ThumbnailOptions>,
+	timestamp: number,
+	timeoutMs: number,
+	abortSignal?: AbortSignal,
+	localOnly = false,
+): Promise<Uint8Array> {
+	abortSignal?.throwIfAborted();
 	const qualityValue = Math.max(
 		2,
 		Math.min(31, Math.round(31 - (opts.quality / 100) * 29)),
 	);
 	const ffmpegArgs = [
 		"ffmpeg",
+		...(localOnly ? ["-protocol_whitelist", "file,pipe"] : []),
 		"-ss",
 		timestamp.toString(),
 		"-i",
@@ -1746,12 +1790,16 @@ export async function generateThumbnail(
 				]);
 				abortSignal?.throwIfAborted();
 
+				if (totalBytes === 0) {
+					throw new EmptyThumbnailError(
+						exitCode === 0
+							? "FFmpeg produced empty thumbnail"
+							: `FFmpeg thumbnail exited with code ${exitCode}`,
+					);
+				}
 				if (exitCode !== 0) {
 					throw new Error(`FFmpeg thumbnail exited with code ${exitCode}`);
 				}
-				if (totalBytes === 0) {
-					throw new Error("FFmpeg produced empty thumbnail");
-				}
 
 				const output = new Uint8Array(totalBytes);
 				let offset = 0;
@@ -1762,7 +1810,7 @@ export async function generateThumbnail(
 
 				return output;
 			})(),
-			THUMBNAIL_TIMEOUT_MS,
+			timeoutMs,
 			stop,
 		);
 	} catch (error) {

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

@@ -1352,7 +1352,12 @@ async function processVideoAsync(
 		});
 		await sendWebhook(job);
 
-		await uploadFileToS3(outputTempFile.path, outputPresignedUrl, "video/mp4");
+		await uploadFileToS3(
+			outputTempFile.path,
+			outputPresignedUrl,
+			"video/mp4",
+			abortController.signal,
+		);
 
 		if (thumbnailPresignedUrl || previewGifPresignedUrl) {
 			updateJob(jobId, {
@@ -1367,8 +1372,23 @@ async function processVideoAsync(
 			const thumbnailData = await generateThumbnail(
 				outputTempFile.path,
 				metadata.duration,
+				{},
+				abortController.signal,
 			);
-			await uploadToS3(thumbnailData, thumbnailPresignedUrl, "image/jpeg");
+			try {
+				await uploadToS3(
+					thumbnailData,
+					thumbnailPresignedUrl,
+					"image/jpeg",
+					abortController.signal,
+				);
+			} catch (error) {
+				abortController.signal.throwIfAborted();
+				console.warn(
+					`[video/process] Thumbnail upload failed for ${jobId}:`,
+					error,
+				);
+			}
 		}
 
 		await generateAndUploadPreviewGif(