fix: harden video processing queues and recovery

Author: richiemcilroy

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

@@ -67,6 +67,23 @@ describe("mediaProbe integration tests", () => {
 			);
 		});
 
+		test("rejects missing remote media before starting MediaBunny workers", async () => {
+			const server = Bun.serve({
+				port: 0,
+				fetch() {
+					return new Response("Not found", { status: 404 });
+				},
+			});
+
+			try {
+				await expect(probeVideo(`${server.url}missing.mp4`)).rejects.toThrow(
+					"Media input is not accessible",
+				);
+			} finally {
+				await server.stop(true);
+			}
+		});
+
 		test("probes media when HEAD is forbidden but GET is allowed", async () => {
 			const videoData = readFileSync(TEST_VIDEO_WITH_AUDIO_PATH);
 			const server = Bun.serve({

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

@@ -15,6 +15,7 @@ import { probeVideo } from "../../lib/media-probe";
 import {
 	buildStreamingDownloadFfmpegArgs,
 	copyFileToMp4,
+	estimateMaterializedStreamingDurationSeconds,
 	generatePreviewGif,
 	generateThumbnail,
 	getFfmpegHlsCapabilities,
@@ -839,6 +840,42 @@ describe("processVideo integration tests", () => {
 });
 
 describe("ffmpeg-backed media utilities integration tests", () => {
+	test("estimates streaming duration from local manifests without probing remote segments", async () => {
+		const workDir = mkdtempSync(join(tmpdir(), "cap-manifest-duration-"));
+		try {
+			writeFileSync(
+				join(workDir, "video.m3u8"),
+				"#EXTM3U\n#EXTINF:10.5,\nvideo-1.ts\n#EXTINF:8.25,\nvideo-2.ts\n",
+			);
+			writeFileSync(
+				join(workDir, "audio.m3u8"),
+				"#EXTM3U\n#EXTINF:7.0,\naudio-1.ts\n#EXTINF:8.0,\naudio-2.ts\n",
+			);
+
+			expect(await estimateMaterializedStreamingDurationSeconds(workDir)).toBe(
+				18.75,
+			);
+		} finally {
+			rmSync(workDir, { recursive: true, force: true });
+		}
+	});
+
+	test("estimates streaming duration from a DASH presentation attribute", async () => {
+		const workDir = mkdtempSync(join(tmpdir(), "cap-mpd-duration-"));
+		try {
+			writeFileSync(
+				join(workDir, "video.mpd"),
+				'<MPD mediaPresentationDuration="PT1M30.5S"></MPD>',
+			);
+
+			expect(await estimateMaterializedStreamingDurationSeconds(workDir)).toBe(
+				90.5,
+			);
+		} finally {
+			rmSync(workDir, { recursive: true, force: true });
+		}
+	});
+
 	test("uses only legacy HLS options when newer FFmpeg options are unavailable", () => {
 		const capabilities = parseFfmpegHlsCapabilities(`
 			-allowed_extensions <string>

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

@@ -472,6 +472,37 @@ describe("POST /video/process", () => {
 		expect(data.code).toBe("SERVER_BUSY");
 	});
 
+	test("reserves one processing slot for normal-priority recordings", async () => {
+		const resources = jobManager.getSystemResources();
+		mock.module("../../lib/job-manager", () => ({
+			...jobManager,
+			canAcceptNewVideoProcess: () => true,
+			getActiveVideoProcessCount: () => 3,
+			getMaxConcurrentVideoProcesses: () => 4,
+			getSystemResources: () => ({
+				...resources,
+				effectiveMax: 4,
+			}),
+		}));
+
+		const { default: appWithMock } = await import("../../app");
+		const response = await appWithMock.fetch(
+			videoPostRequest("/video/process", {
+				videoId: "bulk-video",
+				userId: "user-id",
+				videoUrl: "https://example.com/video.mp4",
+				outputPresignedUrl: "https://s3.example.com/output",
+				priority: "bulk",
+			}),
+		);
+
+		expect(response.status).toBe(503);
+		expect(response.headers.get("Retry-After")).toBe("15");
+		const data = await response.json();
+		expect(data.code).toBe("SERVER_BUSY");
+		expect(data.activeVideoProcesses).toBe(3);
+	});
+
 	test("returns jobId when process starts successfully", async () => {
 		mock.module("../../lib/job-manager", () => ({
 			canAcceptNewVideoProcess: () => true,

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

@@ -47,11 +47,12 @@ function isHttpUrl(path: string): boolean {
 
 async function hasHttpNetworkFailure(path: string): Promise<boolean> {
 	try {
-		await probeFetch(path, {
-			method: "HEAD",
+		const response = await probeFetch(path, {
+			headers: { Range: "bytes=0-0" },
 			signal: AbortSignal.timeout(10_000),
 		});
-		return false;
+		await response.body?.cancel();
+		return !response.ok;
 	} catch {
 		return true;
 	}

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

@@ -4,7 +4,6 @@ import { join } from "node:path";
 import { type BunFile, file, spawn } from "bun";
 import type { VideoMetadata } from "./job-manager";
 import {
-	createMediaInput,
 	DOWNLOAD_TIMEOUT_MS,
 	PROCESS_TIMEOUT_MS,
 	type ProgressCallback,
@@ -28,7 +27,6 @@ const MAX_PROCESS_TIMEOUT_MS = 2 * 60 * 60 * 1000;
 // long (e.g. 60+ minute) manifest with 1000+ segments.
 const STREAMING_DOWNLOAD_TIMEOUT_PER_SECOND_MS = 5_000;
 const MAX_STREAMING_DOWNLOAD_TIMEOUT_MS = 30 * 60 * 1000;
-const STREAMING_DURATION_PROBE_TIMEOUT_MS = 15_000;
 const THUMBNAIL_TIMEOUT_MS = 60_000;
 const PREVIEW_GIF_TIMEOUT_MS = 30_000;
 const PROBE_H264_LEVEL_TIMEOUT_MS = 10_000;
@@ -1117,24 +1115,35 @@ export function buildStreamingDownloadFfmpegArgs(
 	];
 }
 
-async function probeStreamingDurationSeconds(
-	videoUrl: string,
+export async function estimateMaterializedStreamingDurationSeconds(
+	dirPath: string,
 ): Promise<number | null> {
-	try {
-		const input = createMediaInput(videoUrl);
-		try {
-			const duration = await withTimeout(
-				input.computeDuration(),
-				STREAMING_DURATION_PROBE_TIMEOUT_MS,
-			);
-			return Number.isFinite(duration) && duration > 0 ? duration : null;
-		} finally {
-			input.dispose();
+	let longestDuration = 0;
+
+	for (const entry of await readdir(dirPath)) {
+		if (!entry.endsWith(".m3u8") && !entry.endsWith(".mpd")) continue;
+		const content = await file(join(dirPath, entry)).text();
+
+		if (entry.endsWith(".mpd")) {
+			const durationAttribute = content.match(
+				/\bmediaPresentationDuration\s*=\s*["']([^"']+)["']/i,
+			)?.[1];
+			const duration = parseIsoDurationSeconds(durationAttribute);
+			if (duration) longestDuration = Math.max(longestDuration, duration);
+			continue;
 		}
-	} catch {
-		// Best-effort: fall back to the flat DOWNLOAD_TIMEOUT_MS budget below.
-		return null;
+
+		let playlistDuration = 0;
+		for (const match of content.matchAll(/^#EXTINF:([\d.]+)/gm)) {
+			const segmentDuration = Number(match[1]);
+			if (Number.isFinite(segmentDuration) && segmentDuration > 0) {
+				playlistDuration += segmentDuration;
+			}
+		}
+		longestDuration = Math.max(longestDuration, playlistDuration);
 	}
+
+	return longestDuration > 0 ? longestDuration : null;
 }
 
 function getStreamingDownloadTimeoutMs(durationSeconds: number | null): number {
@@ -1168,7 +1177,8 @@ async function downloadStreamingVideoToTemp(
 			manifestDir,
 			abortSignal,
 		);
-		const durationSeconds = await probeStreamingDurationSeconds(videoUrl);
+		const durationSeconds =
+			await estimateMaterializedStreamingDurationSeconds(manifestDir);
 		const downloadTimeoutMs = getStreamingDownloadTimeoutMs(durationSeconds);
 
 		await runFfmpegCommand(

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

@@ -89,6 +89,7 @@ const processSchema = z.object({
 	webhookUrl: z.string().url().optional(),
 	webhookSecret: z.string().optional(),
 	inputExtension: z.string().optional(),
+	priority: z.enum(["normal", "bulk"]).optional(),
 	maxWidth: z.number().max(4096).optional(),
 	maxHeight: z.number().max(4096).optional(),
 	crf: z.number().min(0).max(51).optional(),
@@ -652,9 +653,14 @@ video.post("/process", async (c) => {
 		);
 	}
 
-	if (!canAcceptNewVideoProcess()) {
+	const capacity = getVideoCapacitySnapshot();
+	const bulkCapacityReached =
+		result.data.priority === "bulk" &&
+		capacity.effectiveMaxVideoProcesses > 1 &&
+		capacity.activeVideoProcesses >= capacity.effectiveMaxVideoProcesses - 1;
+	if (bulkCapacityReached || !canAcceptNewVideoProcess()) {
 		c.header("Retry-After", VIDEO_BUSY_RETRY_AFTER_SECONDS.toString());
-		return c.json(getBusyResponseBody(getVideoCapacitySnapshot()), 503);
+		return c.json(getBusyResponseBody(capacity), 503);
 	}
 
 	const {
@@ -1088,10 +1094,8 @@ async function editVideoAsync(
 			await sendWebhook(downloadingJob);
 		}
 
-		const inputTempFile = await downloadVideoToTemp(
-			sourceUrl,
-			".mp4",
-			abortController.signal,
+		const inputTempFile = await withJobHeartbeat(jobId, () =>
+			downloadVideoToTemp(sourceUrl, ".mp4", abortController.signal),
 		);
 		updateJob(jobId, { inputTempFile });
 
@@ -1254,10 +1258,12 @@ async function processVideoAsync(
 		});
 		await sendWebhook(job);
 
-		const inputTempFile = await downloadVideoToTemp(
-			videoUrl,
-			options.inputExtension,
-			abortController.signal,
+		const inputTempFile = await withJobHeartbeat(jobId, () =>
+			downloadVideoToTemp(
+				videoUrl,
+				options.inputExtension,
+				abortController.signal,
+			),
 		);
 		updateJob(jobId, { inputTempFile });
 

apps/web/__tests__/integration/transcribe-workflow.test.ts

@@ -245,6 +245,53 @@ describe("transcribeVideoWorkflow", () => {
 		expect(mocks.updates.at(-1)).toEqual({ transcriptionStatus: "COMPLETE" });
 	});
 
+	it("marks audio without speech as skipped without retrying transcription", async () => {
+		mocks.transcribe.mockResolvedValueOnce({
+			id: "silent-transcript",
+			status: "error",
+			error:
+				"language_detection cannot be performed on files with no spoken audio.",
+		});
+
+		const { transcribeVideoWorkflow } = await import("@/workflows/transcribe");
+		const result = await transcribeVideoWorkflow({
+			videoId: "video-123",
+			userId: "user-456",
+			aiGenerationEnabled: true,
+		});
+
+		expect(result).toEqual({
+			success: true,
+			message: "Video has no spoken audio - skipped transcription",
+		});
+		expect(mocks.transcribe).toHaveBeenCalledTimes(1);
+		expect(mocks.updates).toContainEqual({ transcriptionStatus: "NO_AUDIO" });
+		expect(mocks.updates).not.toContainEqual({ transcriptionStatus: "ERROR" });
+		expect(mocks.startAiGeneration).not.toHaveBeenCalled();
+	});
+
+	it("preserves transcription failures unrelated to missing speech", async () => {
+		mocks.transcribe.mockResolvedValueOnce({
+			id: "failed-transcript",
+			status: "error",
+			error: "Audio could not be decoded",
+		});
+
+		const { transcribeVideoWorkflow } = await import("@/workflows/transcribe");
+
+		await expect(
+			transcribeVideoWorkflow({
+				videoId: "video-123",
+				userId: "user-456",
+				aiGenerationEnabled: false,
+			}),
+		).rejects.toThrow("Audio could not be decoded");
+		expect(mocks.updates).toContainEqual({ transcriptionStatus: "ERROR" });
+		expect(mocks.updates).not.toContainEqual({
+			transcriptionStatus: "NO_AUDIO",
+		});
+	});
+
 	it("never overwrites the original-timeline transcript of an edited video", async () => {
 		state.editRows = [
 			{

apps/web/__tests__/unit/media-server-backpressure.test.ts

@@ -0,0 +1,63 @@
+import { afterEach, describe, expect, it, vi } from "vitest";
+import { RetryableError } from "workflow";
+import {
+	createMediaServerCapacityError,
+	isMediaServerCapacityError,
+} from "@/lib/media-server-backpressure";
+
+afterEach(() => {
+	vi.useRealTimers();
+});
+
+describe("media server backpressure", () => {
+	it("recognizes durable workflow errors caused by exhausted media capacity", () => {
+		expect(isMediaServerCapacityError(new Error("Server is busy"))).toBe(true);
+		expect(isMediaServerCapacityError("SERVER_BUSY: at capacity")).toBe(true);
+		expect(isMediaServerCapacityError(new Error("Invalid video source"))).toBe(
+			false,
+		);
+	});
+
+	it("schedules durable retries after the server's requested delay", () => {
+		vi.useFakeTimers();
+		vi.setSystemTime(new Date("2026-08-25T12:00:00.000Z"));
+		const error = createMediaServerCapacityError({
+			response: new Response(null, {
+				status: 503,
+				headers: { "Retry-After": "15" },
+			}),
+			message: "Server is busy",
+			videoId: "video-1",
+		});
+
+		expect(error).toBeInstanceOf(RetryableError);
+		const retryDelay = new Date(error.retryAfter).getTime() - Date.now();
+		expect(retryDelay).toBeGreaterThanOrEqual(15_000);
+		expect(retryDelay).toBeLessThan(35_000);
+	});
+
+	it("gives ordinary recordings priority over bulk imports", () => {
+		vi.useFakeTimers();
+		vi.setSystemTime(new Date("2026-08-25T12:00:00.000Z"));
+		const response = new Response(null, {
+			status: 503,
+			headers: { "Retry-After": "15" },
+		});
+		const recording = createMediaServerCapacityError({
+			response,
+			message: "Server is busy",
+			videoId: "video-1",
+		});
+		const bulk = createMediaServerCapacityError({
+			response,
+			message: "Server is busy",
+			videoId: "video-1",
+			priority: "bulk",
+		});
+
+		expect(
+			new Date(bulk.retryAfter).getTime() -
+				new Date(recording.retryAfter).getTime(),
+		).toBe(15_000);
+	});
+});