From f139bfe93ef8f24abd77b0dad73e20228185fcf6 Mon Sep 17 00:00:00 2001
From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com>
Date: Sun, 6 Sep 2026 20:29:22 +0100
Subject: [PATCH 1/6] fix: exclude reclaimable file cache from recording memory
 pressure

---
 .../__tests__/lib/container-memory.test.ts    | 57 +++++++++++++++++
 ...edia-routes-real-world.integration.test.ts |  2 +
 apps/media-server/src/lib/container-memory.ts | 63 ++++++++++++++++---
 apps/media-server/src/lib/job-manager.ts      |  6 +-
 4 files changed, 118 insertions(+), 10 deletions(-)

diff --git a/apps/media-server/src/__tests__/lib/container-memory.test.ts b/apps/media-server/src/__tests__/lib/container-memory.test.ts
index 71e215e80be..0fd61e02251 100644
--- a/apps/media-server/src/__tests__/lib/container-memory.test.ts
+++ b/apps/media-server/src/__tests__/lib/container-memory.test.ts
@@ -51,3 +51,60 @@ describe("container memory metrics", () => {
 		expect(metrics.pressure).toBe(0.75);
 	});
 });
+
+test.each(["memory.current", "memory.usage_in_bytes"])(
+	"excludes only clean inactive cache from %s pressure",
+	async (filename) => {
+		const dir = await mkdtemp(join(tmpdir(), "cap-container-cache-"));
+		tempDirs.push(dir);
+		const usagePath = join(dir, filename);
+		await writeFile(usagePath, String(1000 * 1024 ** 2));
+		const prefix = filename === "memory.current" ? "" : "total_";
+		await writeFile(
+			join(dir, "memory.stat"),
+			[
+				`${prefix}inactive_file ${800 * 1024 ** 2}`,
+				`${prefix}${prefix ? "dirty" : "file_dirty"} ${50 * 1024 ** 2}`,
+				`${prefix}${prefix ? "writeback" : "file_writeback"} ${25 * 1024 ** 2}`,
+			].join("\n"),
+		);
+		expect(
+			getContainerMemoryMetrics({
+				usagePaths: [usagePath],
+				configuredLimitMB: 1024,
+			}),
+		).toMatchObject({
+			usageMB: 1000,
+			workingSetMB: 275,
+			reclaimableCacheMB: 725,
+			pressure: 275 / 1024,
+		});
+	},
+);
+
+test.each([
+	"",
+	"inactive_file invalid",
+	"inactive_file -1",
+	"inactive_file 999999999999999999999",
+	"inactive_file 100\nfile_dirty 0",
+])(
+	"retains conservative pressure when cache accounting is invalid (%s)",
+	async (stats) => {
+		const dir = await mkdtemp(join(tmpdir(), "cap-container-cache-"));
+		tempDirs.push(dir);
+		const usagePath = join(dir, "memory.current");
+		await writeFile(usagePath, String(950 * 1024 ** 2));
+		await writeFile(join(dir, "memory.stat"), stats);
+		expect(
+			getContainerMemoryMetrics({
+				usagePaths: [usagePath],
+				configuredLimitMB: 1000,
+			}),
+		).toMatchObject({
+			workingSetMB: 950,
+			reclaimableCacheMB: 0,
+			pressure: 0.95,
+		});
+	},
+);
diff --git a/apps/media-server/src/__tests__/lib/media-routes-real-world.integration.test.ts b/apps/media-server/src/__tests__/lib/media-routes-real-world.integration.test.ts
index 50f5a5c275f..49cfa781237 100644
--- a/apps/media-server/src/__tests__/lib/media-routes-real-world.integration.test.ts
+++ b/apps/media-server/src/__tests__/lib/media-routes-real-world.integration.test.ts
@@ -434,6 +434,8 @@ beforeEach(() => {
 	spyOn(containerCpu, "getContainerCpuUsageMicros").mockReturnValue(0);
 	spyOn(containerMemory, "getContainerMemoryMetrics").mockReturnValue({
 		usageMB: 256,
+		workingSetMB: 256,
+		reclaimableCacheMB: 0,
 		limitMB: 4096,
 		pressure: 0.0625,
 	});
diff --git a/apps/media-server/src/lib/container-memory.ts b/apps/media-server/src/lib/container-memory.ts
index 22633dd2cca..c008faf11ba 100644
--- a/apps/media-server/src/lib/container-memory.ts
+++ b/apps/media-server/src/lib/container-memory.ts
@@ -1,4 +1,5 @@
 import { existsSync, readFileSync } from "node:fs";
+import { dirname, join } from "node:path";
 
 const CGROUP_MEMORY_LIMIT_PATHS = [
 	"/sys/fs/cgroup/memory.max",
@@ -10,7 +11,7 @@ const CGROUP_MEMORY_USAGE_PATHS = [
 ];
 const MAX_PLAUSIBLE_CONTAINER_LIMIT_BYTES = 1024 ** 5;
 
-function readMemoryValueMB(paths: string[], enforcePlausibleLimit: boolean) {
+function readMemoryValue(paths: string[], enforcePlausibleLimit: boolean) {
 	for (const path of paths) {
 		if (!existsSync(path)) continue;
 
@@ -29,15 +30,17 @@ function readMemoryValueMB(paths: string[], enforcePlausibleLimit: boolean) {
 			bytes > 0 &&
 			(!enforcePlausibleLimit || bytes < MAX_PLAUSIBLE_CONTAINER_LIMIT_BYTES)
 		) {
-			return Math.round(bytes / (1024 * 1024));
+			return { bytes, path };
 		}
 	}
 
-	return 0;
+	return undefined;
 }
 
 export interface ContainerMemoryMetrics {
 	usageMB: number;
+	workingSetMB: number;
+	reclaimableCacheMB: number;
 	limitMB: number;
 	pressure: number;
 }
@@ -54,17 +57,59 @@ export function getContainerMemoryMetrics(
 	const configuredLimitMB =
 		options.configuredLimitMB ??
 		(Number.parseInt(process.env.MEDIA_SERVER_MEMORY_LIMIT_MB ?? "0", 10) || 0);
-	const limitMB =
-		configuredLimitMB ||
-		readMemoryValueMB(options.limitPaths ?? CGROUP_MEMORY_LIMIT_PATHS, true);
-	const usageMB = readMemoryValueMB(
+	const limit = readMemoryValue(
+		options.limitPaths ?? CGROUP_MEMORY_LIMIT_PATHS,
+		true,
+	);
+	const limitMB = configuredLimitMB || (limit ? limit.bytes / 1024 ** 2 : 0);
+	const usage = readMemoryValue(
 		options.usagePaths ?? CGROUP_MEMORY_USAGE_PATHS,
 		false,
 	);
-
+	const usageBytes = usage?.bytes ?? 0;
+	let reclaimableBytes = 0;
+	if (usage) {
+		try {
+			const stats = new Map(
+				readFileSync(join(dirname(usage.path), "memory.stat"), "utf8")
+					.trim()
+					.split("\n")
+					.map((line) => {
+						const [key, value] = line.trim().split(/\s+/);
+						return [key, Number(value)] as const;
+					}),
+			);
+			const prefix = usage.path.endsWith("memory.usage_in_bytes")
+				? "total_"
+				: "";
+			const inactive = stats.get(`${prefix}inactive_file`);
+			const dirty = stats.get(`${prefix}${prefix ? "dirty" : "file_dirty"}`);
+			const writeback = stats.get(
+				`${prefix}${prefix ? "writeback" : "file_writeback"}`,
+			);
+			if (
+				[inactive, dirty, writeback].every(
+					(value) =>
+						typeof value === "number" &&
+						Number.isSafeInteger(value) &&
+						value >= 0,
+				)
+			) {
+				reclaimableBytes = Math.min(
+					usageBytes,
+					Math.max(0, (inactive ?? 0) - (dirty ?? 0) - (writeback ?? 0)),
+				);
+			}
+		} catch {}
+	}
+	const usageMB = usageBytes / 1024 ** 2;
+	const reclaimableCacheMB = reclaimableBytes / 1024 ** 2;
+	const workingSetMB = usageMB - reclaimableCacheMB;
 	return {
 		usageMB,
+		workingSetMB,
+		reclaimableCacheMB,
 		limitMB,
-		pressure: limitMB > 0 && usageMB > 0 ? usageMB / limitMB : 0,
+		pressure: limitMB > 0 ? workingSetMB / limitMB : 0,
 	};
 }
diff --git a/apps/media-server/src/lib/job-manager.ts b/apps/media-server/src/lib/job-manager.ts
index 88232e13767..74caa0deaa7 100644
--- a/apps/media-server/src/lib/job-manager.ts
+++ b/apps/media-server/src/lib/job-manager.ts
@@ -259,6 +259,8 @@ export interface SystemResources {
 	processHeapMB: number;
 	processRssLimitMB: number;
 	containerMemoryUsageMB: number;
+	containerMemoryWorkingSetMB: number;
+	containerMemoryReclaimableCacheMB: number;
 	containerMemoryLimitMB: number;
 	memoryPressure: number;
 	configuredMax: number;
@@ -274,7 +276,7 @@ export function getSystemResources(): SystemResources {
 	const processRssMB = Math.round(mem.rss / (1024 * 1024));
 	const processHeapMB = Math.round(mem.heapUsed / (1024 * 1024));
 	const containerMemory = getContainerMemoryMetrics();
-	const memoryUsageMB = containerMemory.usageMB || processRssMB;
+	const memoryUsageMB = Math.max(containerMemory.workingSetMB, processRssMB);
 	const memoryLimitMB = containerMemory.limitMB;
 	const memoryPressure = memoryLimitMB > 0 ? memoryUsageMB / memoryLimitMB : 0;
 	const max = getMaxConcurrentVideoProcesses();
@@ -313,6 +315,8 @@ export function getSystemResources(): SystemResources {
 		processHeapMB,
 		processRssLimitMB: memoryLimitMB,
 		containerMemoryUsageMB: containerMemory.usageMB,
+		containerMemoryWorkingSetMB: containerMemory.workingSetMB,
+		containerMemoryReclaimableCacheMB: containerMemory.reclaimableCacheMB,
 		containerMemoryLimitMB: containerMemory.limitMB,
 		memoryPressure,
 		configuredMax: configuredMaxProcesses,

From bc53298e2459beb5f19e2ad0160b9e80cf874621 Mon Sep 17 00:00:00 2001
From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com>
Date: Sun, 6 Sep 2026 20:29:22 +0100
Subject: [PATCH 2/6] fix: resume large Drive outputs in bounded upload chunks

---
 .../workflows/docker-build-media-server.yml   |   8 +-
 apps/media-server/Dockerfile                  |   6 +-
 .../lib/drive-resumable-upload.test.ts        | 225 ++++++++++++++++++
 .../src/lib/drive-resumable-upload.ts         | 101 ++++++++
 apps/media-server/src/lib/media-video.ts      |  21 +-
 5 files changed, 353 insertions(+), 8 deletions(-)
 create mode 100644 apps/media-server/src/__tests__/lib/drive-resumable-upload.test.ts
 create mode 100644 apps/media-server/src/lib/drive-resumable-upload.ts

diff --git a/.github/workflows/docker-build-media-server.yml b/.github/workflows/docker-build-media-server.yml
index 74cd288462f..1925be2822a 100644
--- a/.github/workflows/docker-build-media-server.yml
+++ b/.github/workflows/docker-build-media-server.yml
@@ -84,7 +84,10 @@ jobs:
             src/__tests__/lib/media-size.test.ts \
             src/__tests__/lib/media-probe.integration.test.ts
           docker run --rm --network none --entrypoint bun "$MEDIA_IMAGE" test \
-            src/__tests__/lib/media-transfer.test.ts
+            src/__tests__/lib/media-transfer.test.ts \
+            src/__tests__/lib/drive-resumable-upload.test.ts \
+            src/__tests__/lib/storage-upload.test.ts \
+            src/__tests__/lib/container-memory.test.ts
           docker run --rm --network none --entrypoint bun "$MEDIA_IMAGE" test \
             src/__tests__/routes/recording-verification.test.ts
           docker run --rm --network none --entrypoint bun "$MEDIA_IMAGE" test \
@@ -106,7 +109,8 @@ jobs:
           docker run --rm --network none --cpus 2 --memory 2g \
             -e MEDIA_SERVER_TRANSFER_PERFORMANCE_TESTS=1 \
             --entrypoint bun "$MEDIA_IMAGE" test \
-            src/__tests__/lib/media-transfer.integration.test.ts
+            src/__tests__/lib/media-transfer.integration.test.ts \
+            src/__tests__/lib/drive-resumable-upload.test.ts
 
       - name: Export Digest
         if: github.event_name != 'pull_request'
diff --git a/apps/media-server/Dockerfile b/apps/media-server/Dockerfile
index 2c77c9ca6b1..04f8d241bd7 100644
--- a/apps/media-server/Dockerfile
+++ b/apps/media-server/Dockerfile
@@ -11,7 +11,8 @@ RUN bun install --frozen-lockfile --production
 
 COPY apps/media-server/src ./src
 
-RUN bun test src/__tests__/lib/media-size.test.ts src/__tests__/lib/media-probe.integration.test.ts \
+RUN bun test src/__tests__/lib/drive-resumable-upload.test.ts src/__tests__/lib/storage-upload.test.ts src/__tests__/lib/container-memory.test.ts \
+	&& bun test src/__tests__/lib/media-size.test.ts src/__tests__/lib/media-probe.integration.test.ts \
 	&& bun test src/__tests__/lib/recording-verification.integration.test.ts src/__tests__/lib/job-manager.test.ts \
 	&& bun test src/__tests__/lib/media-transfer.test.ts \
 	&& bun test src/__tests__/routes/recording-verification.test.ts \
@@ -24,7 +25,8 @@ RUN MEDIA_SERVER_RECORDING_PERFORMANCE_TESTS=1 bun test \
 	--test-name-pattern 'streams a complete long recording'
 
 RUN MEDIA_SERVER_TRANSFER_PERFORMANCE_TESTS=1 bun test \
-	src/__tests__/lib/media-transfer.integration.test.ts
+	src/__tests__/lib/media-transfer.integration.test.ts \
+	src/__tests__/lib/drive-resumable-upload.test.ts
 
 ENV PORT=3456
 EXPOSE 3456
diff --git a/apps/media-server/src/__tests__/lib/drive-resumable-upload.test.ts b/apps/media-server/src/__tests__/lib/drive-resumable-upload.test.ts
new file mode 100644
index 00000000000..ed9660b3ff3
--- /dev/null
+++ b/apps/media-server/src/__tests__/lib/drive-resumable-upload.test.ts
@@ -0,0 +1,225 @@
+import { afterEach, expect, test } from "bun:test";
+import { mkdtemp, open, rm } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { uploadDriveResumable } from "../../lib/drive-resumable-upload";
+
+const originalFetch = globalThis.fetch;
+const chunk = 32 * 1024 ** 2;
+const dirs: string[] = [];
+afterEach(async () => {
+	globalThis.fetch = originalFetch;
+	await Promise.all(
+		dirs.splice(0).map((dir) => rm(dir, { recursive: true, force: true })),
+	);
+});
+
+async function sparseFile(size: number) {
+	const dir = await mkdtemp(join(tmpdir(), "cap-drive-upload-"));
+	dirs.push(dir);
+	const path = join(dir, "output.mp4");
+	const handle = await open(path, "w");
+	try {
+		await handle.truncate(size);
+		await handle.write(new Uint8Array([99]), 0, 1, size - 1);
+	} finally {
+		await handle.close();
+	}
+	return Bun.file(path);
+}
+
+function incomplete(end?: number) {
+	return new Response(null, {
+		status: 308,
+		headers: end === undefined ? {} : { Range: `bytes=0-${end - 1}` },
+	});
+}
+
+test("slices real files beyond 8 GiB without truncating offsets or allocating the file", async () => {
+	const size = 9 * 1024 ** 3 + 123;
+	const body = await sparseFile(size);
+	let received = 0;
+	globalThis.fetch = (async (_url, init) => {
+		const piece = init?.body as Blob;
+		const end = Math.min(received + chunk, size);
+		expect(piece.size).toBe(end - received);
+		expect(new Headers(init?.headers).get("Content-Range")).toBe(
+			`bytes ${received}-${end - 1}/${size}`,
+		);
+		if (end === size)
+			expect(new Uint8Array(await piece.slice(-1).arrayBuffer())[0]).toBe(99);
+		received = end;
+		return received === size
+			? Response.json({ size: String(size) })
+			: incomplete(received);
+	}) as typeof fetch;
+	const result = await uploadDriveResumable(
+		"https://drive.test/upload",
+		body,
+		"video/mp4",
+	);
+	expect(received).toBe(size);
+	expect(await result.json()).toEqual({ size: String(size) });
+});
+
+test.each(["disconnect", "503"])(
+	"resumes confirmed partial bytes after %s without replaying the prefix",
+	async (mode) => {
+		const body = await sparseFile(chunk + 123);
+		const requests: string[] = [];
+		globalThis.fetch = (async (_url, init) => {
+			requests.push(new Headers(init?.headers).get("Content-Range") ?? "");
+			if (requests.length === 1) {
+				if (mode === "disconnect") throw new Error("Connection lost");
+				return new Response(null, { status: 503 });
+			}
+			if (requests.length === 2) return incomplete(chunk / 2);
+			expect((init?.body as Blob).size).toBe(chunk / 2 + 123);
+			return Response.json({ done: true });
+		}) as typeof fetch;
+		await uploadDriveResumable("https://drive.test/upload", body, "video/mp4");
+		expect(requests).toEqual([
+			`bytes 0-${chunk - 1}/${body.size}`,
+			`bytes */${body.size}`,
+			`bytes ${chunk / 2}-${body.size - 1}/${body.size}`,
+		]);
+	},
+);
+
+test("reconciles a lost final response without sending the output again", async () => {
+	const requests: string[] = [];
+	globalThis.fetch = (async (_url, init) => {
+		requests.push(new Headers(init?.headers).get("Content-Range") ?? "");
+		if (requests.length === 1) throw new Error("Final response lost");
+		return Response.json({ done: true });
+	}) as typeof fetch;
+	await uploadDriveResumable(
+		"https://drive.test/upload",
+		new Blob(["abc"]),
+		"video/mp4",
+	);
+	expect(requests).toEqual(["bytes 0-2/3", "bytes */3"]);
+});
+
+test.each(["bytes=0-999", "bytes=1-2", "nonsense", "bytes=0-9007199254740993"])(
+	"rejects untrustworthy offsets %s",
+	async (range) => {
+		globalThis.fetch = (async () =>
+			new Response(null, {
+				status: 308,
+				headers: { Range: range },
+			})) as typeof fetch;
+		await expect(
+			uploadDriveResumable(
+				"https://drive.test/upload",
+				new Blob(["abc"]),
+				"video/mp4",
+			),
+		).rejects.toThrow("invalid upload offset");
+	},
+);
+
+test("bounds an upload that never acknowledges progress", async () => {
+	let requests = 0;
+	globalThis.fetch = (async () => {
+		requests++;
+		return incomplete();
+	}) as typeof fetch;
+	await expect(
+		uploadDriveResumable(
+			"https://drive.test/upload",
+			new Blob(["abc"]),
+			"video/mp4",
+		),
+	).rejects.toThrow("no progress");
+	expect(requests).toBe(9);
+}, 10000);
+
+test("cancels backoff without resending media", async () => {
+	const controller = new AbortController();
+	let requests = 0;
+	globalThis.fetch = (async () => {
+		requests++;
+		controller.abort();
+		throw new Error("lost");
+	}) as typeof fetch;
+	await expect(
+		uploadDriveResumable(
+			"https://drive.test/upload",
+			new Blob(["abc"]),
+			"video/mp4",
+			undefined,
+			controller.signal,
+		),
+	).rejects.toThrow();
+	expect(requests).toBe(1);
+});
+
+test.skipIf(process.env.MEDIA_SERVER_TRANSFER_PERFORMANCE_TESTS !== "1")(
+	"streams a 9 GiB output over HTTP within bounded memory",
+	async () => {
+		const size = 9 * 1024 ** 3 + 123;
+		const body = await sparseFile(size);
+		let received = 0;
+		let lastByte = 0;
+		let requests = 0;
+		const baseline = process.memoryUsage().rss;
+		let peak = baseline;
+		const started = performance.now();
+		const server = Bun.serve({
+			port: 0,
+			hostname: "127.0.0.1",
+			async fetch(request) {
+				requests++;
+				const start = received;
+				const reader = request.body?.getReader();
+				if (!reader) {
+					console.log(
+						JSON.stringify({
+							event: "drive_status_probe",
+							received,
+							range: request.headers.get("Content-Range"),
+						}),
+					);
+					return incomplete(received || undefined);
+				}
+				for (;;) {
+					const next = await reader.read();
+					if (next.done) break;
+					received += next.value.length;
+					lastByte = next.value[next.value.length - 1];
+				}
+				peak = Math.max(peak, process.memoryUsage().rss);
+				expect(request.headers.get("Content-Range")).toBe(
+					`bytes ${start}-${received - 1}/${size}`,
+				);
+				return received === size
+					? Response.json({ done: true })
+					: incomplete(received);
+			},
+		});
+		try {
+			await uploadDriveResumable(
+				`http://127.0.0.1:${server.port}/upload`,
+				body,
+				"video/mp4",
+			);
+			expect(received).toBe(size);
+			expect(lastByte).toBe(99);
+			expect(requests).toBe(Math.ceil(size / chunk));
+			expect(peak - baseline).toBeLessThan(384 * 1024 ** 2);
+			console.log(
+				JSON.stringify({
+					event: "drive_large_upload_verified",
+					bytes: received,
+					requests,
+					elapsedMs: Math.round(performance.now() - started),
+					peakRssGrowthMiB: Math.round((peak - baseline) / 1024 ** 2),
+				}),
+			);
+		} finally {
+			await server.stop(true);
+		}
+	},
+	120000,
+);
diff --git a/apps/media-server/src/lib/drive-resumable-upload.ts b/apps/media-server/src/lib/drive-resumable-upload.ts
new file mode 100644
index 00000000000..8a04d55e805
--- /dev/null
+++ b/apps/media-server/src/lib/drive-resumable-upload.ts
@@ -0,0 +1,101 @@
+import { setTimeout as sleep } from "node:timers/promises";
+import { UPLOAD_TIMEOUT_MS } from "./media-common";
+
+const CHUNK_BYTES = 32 * 1024 * 1024;
+const MAX_RETRIES = 4;
+const MAX_RETRANSMITTED_BYTES = 16 * CHUNK_BYTES;
+
+export async function uploadDriveResumable(
+	url: string,
+	body: Blob,
+	contentType: string,
+	ifNoneMatch?: "*",
+	abortSignal?: AbortSignal,
+): Promise<Response> {
+	const size = body.size;
+	if (!Number.isSafeInteger(size) || size <= 0)
+		throw new Error("Invalid Drive upload size");
+	let offset = 0;
+	let sentThrough = 0;
+	let transmitted = 0;
+	let failures = 0;
+	let queryStatus = false;
+	for (;;) {
+		abortSignal?.throwIfAborted();
+		const end = Math.min(offset + CHUNK_BYTES, size);
+		const querying = queryStatus;
+		const headers: Record<string, string> = {
+			"Content-Type": contentType,
+			"Content-Length": querying ? "0" : String(end - offset),
+			"Content-Range": querying
+				? `bytes */${size}`
+				: `bytes ${offset}-${end - 1}/${size}`,
+		};
+		if (ifNoneMatch) headers["If-None-Match"] = ifNoneMatch;
+		if (!querying) {
+			transmitted += end - offset;
+			sentThrough = Math.max(sentThrough, end);
+			if (transmitted > size + MAX_RETRANSMITTED_BYTES)
+				throw new Error("Drive upload exceeded its retransmission limit");
+		}
+		let response: Response | undefined;
+		let failure: unknown;
+		try {
+			response = await fetch(url, {
+				method: "PUT",
+				headers,
+				body: querying ? undefined : body.slice(offset, end),
+				redirect: "manual",
+				signal: abortSignal
+					? AbortSignal.any([
+							abortSignal,
+							AbortSignal.timeout(UPLOAD_TIMEOUT_MS),
+						])
+					: AbortSignal.timeout(UPLOAD_TIMEOUT_MS),
+			});
+		} catch (error) {
+			abortSignal?.throwIfAborted();
+			failure = error;
+		}
+		if (response?.status === 200 || response?.status === 201) {
+			if (sentThrough === size) return response;
+			await response.body?.cancel().catch(() => {});
+			throw new Error("Drive completed an upload before all bytes were sent");
+		}
+		if (response?.status === 308) {
+			const range = response.headers.get("range");
+			await response.body?.cancel().catch(() => {});
+			const match = range?.match(/^bytes=0-(\d+)$/i);
+			const nextOffset =
+				range === null ? 0 : match ? Number(match[1]) + 1 : NaN;
+			if (
+				!Number.isSafeInteger(nextOffset) ||
+				nextOffset < offset ||
+				nextOffset > sentThrough
+			)
+				throw new Error("Drive returned an invalid upload offset");
+			if (nextOffset > offset) {
+				offset = nextOffset;
+				failures = 0;
+				queryStatus = offset === size;
+				continue;
+			}
+			if (querying && offset < size) {
+				queryStatus = false;
+				continue;
+			}
+			failure = new Error("Drive upload made no progress");
+		} else if (response) {
+			await response.body?.cancel().catch(() => {});
+			failure = new Error(`Drive upload failed: HTTP ${response.status}`);
+			if (![408, 425, 429, 500, 502, 503, 504].includes(response.status))
+				throw failure;
+		}
+		if (++failures > MAX_RETRIES)
+			throw failure instanceof Error
+				? failure
+				: new Error("Drive upload failed after retries");
+		queryStatus = true;
+		await sleep(250 * 2 ** (failures - 1), undefined, { signal: abortSignal });
+	}
+}
diff --git a/apps/media-server/src/lib/media-video.ts b/apps/media-server/src/lib/media-video.ts
index 022fb3fae62..a43b9a4b36c 100644
--- a/apps/media-server/src/lib/media-video.ts
+++ b/apps/media-server/src/lib/media-video.ts
@@ -3,6 +3,7 @@ import { mkdtemp, readdir, rm, writeFile } from "node:fs/promises";
 import { 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,
@@ -2065,6 +2066,22 @@ async function uploadWithRetry(
 	ifNoneMatch?: "*",
 	abortSignal?: AbortSignal,
 ): Promise<StorageUploadReceipt> {
+	if (isGoogleDriveResumableUrl(presignedUrl) && contentLength > 0) {
+		const response = await uploadDriveResumable(
+			presignedUrl,
+			bodyFactory(),
+			contentType,
+			ifNoneMatch,
+			abortSignal,
+		);
+		const receipt = await readUploadReceipt(
+			response,
+			presignedUrl,
+			contentLength,
+		);
+		abortSignal?.throwIfAborted();
+		return receipt;
+	}
 	let lastError: Error | undefined;
 
 	for (let attempt = 0; attempt <= UPLOAD_MAX_RETRIES; attempt++) {
@@ -2077,10 +2094,6 @@ async function uploadWithRetry(
 				"Content-Length": contentLength.toString(),
 			};
 			if (ifNoneMatch) headers["If-None-Match"] = ifNoneMatch;
-			if (isGoogleDriveResumableUrl(presignedUrl) && contentLength > 0) {
-				headers["Content-Range"] =
-					`bytes 0-${contentLength - 1}/${contentLength}`;
-			}
 
 			response = await fetch(presignedUrl, {
 				method: "PUT",

From 459a07314dfaa6e86209ca6490a08c2d111950c2 Mon Sep 17 00:00:00 2001
From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com>
Date: Sun, 6 Sep 2026 20:29:22 +0100
Subject: [PATCH 3/6] fix: process recordings without a daily transfer
 allowance

---
 .../unit/desktop-recording-jobs.test.ts       |  64 ---------
 .../unit/finalize-desktop-recording.test.ts   |  51 +++-----
 .../unit/media-processing-budget.test.ts      | 121 ++++++++++++++++++
 apps/web/lib/desktop-recording-jobs.ts        |  48 -------
 apps/web/lib/media-processing-budget.ts       |  19 +--
 .../workflows/finalize-desktop-recording.ts   |  32 +++--
 6 files changed, 162 insertions(+), 173 deletions(-)
 create mode 100644 apps/web/__tests__/unit/media-processing-budget.test.ts

diff --git a/apps/web/__tests__/unit/desktop-recording-jobs.test.ts b/apps/web/__tests__/unit/desktop-recording-jobs.test.ts
index 1cd3f06f4e1..5a87cf19ae7 100644
--- a/apps/web/__tests__/unit/desktop-recording-jobs.test.ts
+++ b/apps/web/__tests__/unit/desktop-recording-jobs.test.ts
@@ -5,7 +5,6 @@ import {
 	claimProcessingAttempt,
 	type DesktopRecordingAttemptFence,
 	type DesktopRecordingJob,
-	deferForDailyProcessingBudget,
 	ensureSegmentProcessingJob,
 	getDesktopRecordingRetryDelay,
 	getDesktopRecordingWorkerCheckpoint,
@@ -601,69 +600,6 @@ describe("late verification and source commitment", () => {
 });
 
 describe("retained-source retry policy", () => {
-	it("waits for the next UTC allowance without spending the final attempt", async () => {
-		await createAttempt();
-		Object.assign(getJobRow(), {
-			state: "processing",
-			source,
-			attemptId: "budget-attempt",
-			attemptCount: 5,
-			remoteJobId: null,
-		});
-		const fence = {
-			videoId,
-			generation: String(getJobRow().generation),
-			attemptId: "budget-attempt",
-		};
-		expect(await deferForDailyProcessingBudget({ ...fence, now })).toBe(true);
-		expect(await deferForDailyProcessingBudget({ ...fence, now })).toBe(false);
-		expect(getJobRow()).toMatchObject({
-			state: "retry",
-			attemptCount: 4,
-			source,
-			nextRetryAt: new Date("2026-09-03T00:00:00Z"),
-			leaseExpiresAt: null,
-		});
-		expect(
-			await claimProcessingAttempt({
-				videoId,
-				generation: fence.generation,
-				now,
-			}),
-		).toBeNull();
-		expect(
-			await claimProcessingAttempt({
-				videoId,
-				generation: fence.generation,
-				now: new Date("2026-09-03T00:00:00Z"),
-			}),
-		).toMatchObject({ attemptCount: 5, state: "processing", source });
-	});
-
-	it("does not defer an attempt already owned by a remote worker", async () => {
-		await createAttempt();
-		Object.assign(getJobRow(), {
-			state: "processing",
-			source,
-			attemptId: "owned",
-			attemptCount: 5,
-			remoteJobId: "worker",
-		});
-		expect(
-			await deferForDailyProcessingBudget({
-				videoId,
-				generation: String(getJobRow().generation),
-				attemptId: "owned",
-				now,
-			}),
-		).toBe(false);
-		expect(getJobRow()).toMatchObject({
-			state: "processing",
-			attemptCount: 5,
-			remoteJobId: "worker",
-		});
-	});
-
 	it.each([null, source])(
 		"does not recreate a recording while deletion is pending",
 		async (retainedSource) => {
diff --git a/apps/web/__tests__/unit/finalize-desktop-recording.test.ts b/apps/web/__tests__/unit/finalize-desktop-recording.test.ts
index 5dd648595e4..0e6c9aafb08 100644
--- a/apps/web/__tests__/unit/finalize-desktop-recording.test.ts
+++ b/apps/web/__tests__/unit/finalize-desktop-recording.test.ts
@@ -17,7 +17,6 @@ const mocks = vi.hoisted(() => ({
 	retry: vi.fn(),
 	attach: vi.fn(),
 	defer: vi.fn(),
-	deferBudget: vi.fn(),
 	commitSource: vi.fn(),
 	checkpoint: vi.fn(),
 	saveCheckpoint: vi.fn(),
@@ -34,7 +33,7 @@ const mocks = vi.hoisted(() => ({
 vi.mock("@/lib/media-processing-budget", () => ({
 	reserveMediaProcessingBudget: mocks.reserveBudget,
 	MediaProcessingBudgetError: class extends Error {
-		constructor(readonly scope: "daily" | "recording") {
+		constructor(readonly scope: "recording") {
 			super(scope);
 		}
 	},
@@ -64,7 +63,6 @@ vi.mock("@/lib/desktop-recording-jobs", () => ({
 	scheduleRetry: mocks.retry,
 	attachRemoteJob: mocks.attach,
 	deferWithoutMediaServer: mocks.defer,
-	deferForDailyProcessingBudget: mocks.deferBudget,
 }));
 vi.mock("@/lib/desktop-recording-source", () => ({
 	advanceDesktopRecordingSourceCommit: mocks.commitSource,
@@ -203,17 +201,7 @@ beforeEach(() => {
 		withCurrent({ remoteJobId });
 		return true;
 	});
-	mocks.deferBudget.mockImplementation(async () => {
-		const nextRetryAt = new Date();
-		nextRetryAt.setUTCHours(24, 0, 0, 0);
-		withCurrent({
-			state: "retry",
-			leaseExpiresAt: null,
-			nextRetryAt,
-			attemptCount: (current?.attemptCount ?? 1) - 1,
-		});
-		return true;
-	});
+
 	mocks.defer.mockImplementation(async () => {
 		withCurrent({ state: "queued", leaseExpiresAt: null });
 		return true;
@@ -525,13 +513,12 @@ describe("source commitment and media request compatibility", () => {
 		expect(mocks.fetch).not.toHaveBeenCalled();
 	});
 
-	it("sleeps through exhausted days and resumes processing automatically", async () => {
+	it("waits for capacity using the same attempt instead of exhausting processing retries", async () => {
 		withCurrent({ state: "retry", leaseExpiresAt: null, attemptCount: 4 });
-		for (let day = 0; day < 7; day++) {
-			mocks.reserveBudget.mockRejectedValueOnce(
-				new MediaProcessingBudgetError("daily"),
+		for (let index = 0; index < 8; index++)
+			mocks.fetch.mockResolvedValueOnce(
+				Response.json({ code: "SERVER_BUSY" }, { status: 503 }),
 			);
-		}
 		await expect(
 			finalizeDesktopRecordingWorkflow({
 				videoId,
@@ -539,26 +526,22 @@ describe("source commitment and media request compatibility", () => {
 				generation: fixture.generation,
 			}),
 		).resolves.toMatchObject({ success: true });
-		expect(mocks.deferBudget).toHaveBeenCalledTimes(7);
-		expect(
-			mocks.sleep.mock.calls.filter(([delay]) => delay instanceof Date),
-		).toHaveLength(7);
 		expect(current?.attemptCount).toBe(5);
-		expect(mocks.fetch).toHaveBeenCalledTimes(1);
+		expect(mocks.fetch).toHaveBeenCalledTimes(9);
+		const attempts = mocks.reserveBudget.mock.calls.map(
+			([input]) => input.attemptId,
+		);
+		expect(new Set(attempts).size).toBe(1);
+		expect(mocks.retry).not.toHaveBeenCalled();
 		expect(mocks.blocked).not.toHaveBeenCalled();
 	});
 
-	it("defers daily exhaustion without dispatching or permanently blocking the source", async () => {
-		mocks.reserveBudget.mockRejectedValueOnce(
-			new MediaProcessingBudgetError("daily"),
+	it("does not treat an ambiguous dispatch failure as a capacity refusal", async () => {
+		mocks.fetch.mockResolvedValueOnce(
+			Response.json({ code: "UPSTREAM_FAILURE" }, { status: 503 }),
 		);
-		await expect(startDesktopRecordingJob(fixture)).resolves.toEqual({
-			status: "deferred",
-		});
-		expect(mocks.deferBudget).toHaveBeenCalledWith(fixture);
-		expect(mocks.blocked).not.toHaveBeenCalled();
-		expect(mocks.fetch).not.toHaveBeenCalled();
-		expect(mocks.put).not.toHaveBeenCalled();
+		await expect(startDesktopRecordingJob(fixture)).resolves.toBeUndefined();
+		expect(mocks.heartbeat).not.toHaveBeenCalled();
 	});
 
 	it("blocks an exhausted transfer budget before dispatching media work", async () => {
diff --git a/apps/web/__tests__/unit/media-processing-budget.test.ts b/apps/web/__tests__/unit/media-processing-budget.test.ts
new file mode 100644
index 00000000000..6e58c73abce
--- /dev/null
+++ b/apps/web/__tests__/unit/media-processing-budget.test.ts
@@ -0,0 +1,121 @@
+import { beforeEach, expect, it, vi } from "vitest";
+import {
+	getMediaProcessingReservation,
+	reserveMediaProcessingBudget,
+} from "@/lib/media-processing-budget";
+
+const state = vi.hoisted(() => ({
+	rows: new Map<
+		string,
+		{ id: string; limitBytes: number; reservedBytes: number }
+	>(),
+}));
+vi.mock("@cap/database/schema", () => ({
+	mediaProcessingBudgets: {
+		id: "id",
+		reservedBytes: "reservedBytes",
+		expiresAt: "expiresAt",
+	},
+}));
+vi.mock("drizzle-orm", () => ({
+	asc: (value: unknown) => value,
+	lt: vi.fn(),
+	inArray: (_column: unknown, ids: string[]) => ids,
+	sql: (_strings: TemplateStringsArray, ...values: unknown[]) => values,
+}));
+vi.mock("@cap/database", () => ({
+	db: () => ({
+		transaction: async (run: (tx: unknown) => Promise<number>) =>
+			run({
+				insert: () => ({
+					values: (rows: Array<{ id: string; limitBytes: number }>) => ({
+						onDuplicateKeyUpdate: async () => {
+							for (const row of rows)
+								if (!state.rows.has(row.id))
+									state.rows.set(row.id, { ...row, reservedBytes: 0 });
+						},
+					}),
+				}),
+				select: () => ({
+					from: () => ({
+						where: (ids: string[]) => ({
+							orderBy: () => ({
+								for: async () => ids.map((id) => state.rows.get(id)),
+							}),
+						}),
+					}),
+				}),
+				update: () => ({
+					set: (values: { reservedBytes: [unknown, number] }) => ({
+						where: async (ids: string[]) => {
+							for (const id of ids) {
+								const row = state.rows.get(id);
+								if (row) row.reservedBytes += values.reservedBytes[1];
+							}
+						},
+					}),
+				}),
+			}),
+	}),
+}));
+beforeEach(() => state.rows.clear());
+
+it("processes unrelated recordings regardless of the former daily allowance", async () => {
+	vi.stubEnv("MEDIA_PROCESSING_DAILY_BUDGET_GIB", "1");
+	try {
+		for (let index = 0; index < 10; index++) {
+			await expect(
+				reserveMediaProcessingBudget({
+					videoId: `video-${index}`,
+					generation: "generation",
+					attemptId: "attempt",
+					sourceBytes: 1024 ** 3,
+					now: new Date("2026-09-06T23:59:59Z"),
+				}),
+			).resolves.toBe(getMediaProcessingReservation(1024 ** 3).attemptBytes);
+		}
+		expect(state.rows.size).toBe(20);
+	} finally {
+		vi.unstubAllEnvs();
+	}
+});
+
+it("reuses an attempt reservation across dispatch retries and midnight", async () => {
+	const input = {
+		videoId: "video",
+		generation: "generation",
+		attemptId: "attempt",
+		sourceBytes: 1024 ** 3,
+	};
+	const first = await reserveMediaProcessingBudget({
+		...input,
+		now: new Date("2026-09-06T23:59:59Z"),
+	});
+	for (let index = 0; index < 10; index++)
+		expect(
+			await reserveMediaProcessingBudget({
+				...input,
+				now: new Date("2026-09-07T00:00:01Z"),
+			}),
+		).toBe(first);
+	expect([...state.rows.values()].map((row) => row.reservedBytes)).toEqual([
+		first,
+		first,
+	]);
+});
+
+it("still bounds repeated processing of the same immutable source", async () => {
+	const input = {
+		videoId: "video",
+		generation: "generation",
+		sourceBytes: 1024 ** 3,
+	};
+	for (let index = 0; index < 3; index++)
+		await reserveMediaProcessingBudget({
+			...input,
+			attemptId: `attempt-${index}`,
+		});
+	await expect(
+		reserveMediaProcessingBudget({ ...input, attemptId: "attempt-4" }),
+	).rejects.toThrow("recording transfer budget exhausted");
+});
diff --git a/apps/web/lib/desktop-recording-jobs.ts b/apps/web/lib/desktop-recording-jobs.ts
index dd95f644b60..fef2fb35f49 100644
--- a/apps/web/lib/desktop-recording-jobs.ts
+++ b/apps/web/lib/desktop-recording-jobs.ts
@@ -677,54 +677,6 @@ export async function heartbeatAttempt({
 	});
 }
 
-export async function deferForDailyProcessingBudget({
-	now = new Date(),
-	...fence
-}: DesktopRecordingAttemptFence & { now?: Date }): Promise<boolean> {
-	return db().transaction(async (tx) => {
-		const condition = and(
-			attemptCondition(fence),
-			isNull(videoProcessingJobs.remoteJobId),
-		);
-		const [row] = await tx
-			.select()
-			.from(videoProcessingJobs)
-			.where(condition)
-			.for("update");
-		if (
-			!row ||
-			getDesktopRecordingWorkerCheckpoint(parseDesktopRecordingJob(row))
-		)
-			return false;
-		const nextRetryAt = new Date(now);
-		nextRetryAt.setUTCHours(24, 0, 0, 0);
-		await tx
-			.update(videoProcessingJobs)
-			.set({
-				state: "retry",
-				attemptCount: Math.max(0, row.attemptCount - 1),
-				leaseExpiresAt: null,
-				nextRetryAt,
-				errorCode: "processing-daily-budget-exhausted",
-				errorMessage:
-					"Processing will resume when the daily transfer allowance resets.",
-				updatedAt: now,
-			})
-			.where(condition);
-		await tx
-			.update(videoUploads)
-			.set({
-				phase: "processing",
-				processingMessage:
-					"Waiting for processing capacity. Your recording is safely stored.",
-				processingError: null,
-				updatedAt: now,
-			})
-			.where(eq(videoUploads.videoId, fence.videoId));
-		return true;
-	});
-}
-
 export async function scheduleRetry({
 	errorCode,
 	errorMessage,
diff --git a/apps/web/lib/media-processing-budget.ts b/apps/web/lib/media-processing-budget.ts
index 486a2fe3d22..0ee62338aa4 100644
--- a/apps/web/lib/media-processing-budget.ts
+++ b/apps/web/lib/media-processing-budget.ts
@@ -3,10 +3,9 @@ import { db } from "@cap/database";
 import { mediaProcessingBudgets } from "@cap/database/schema";
 import { asc, inArray, lt, sql } from "drizzle-orm";
 
-const GiB = 1024 ** 3;
 const MiB = 1024 ** 2;
 export class MediaProcessingBudgetError extends Error {
-	constructor(readonly scope: "recording" | "daily") {
+	constructor(readonly scope: "recording") {
 		super(`Recording processing paused: ${scope} transfer budget exhausted`);
 		this.name = "MediaProcessingBudgetError";
 	}
@@ -38,15 +37,6 @@ export async function reserveMediaProcessingBudget(input: {
 	const { attemptBytes, recordingBytes } = getMediaProcessingReservation(
 		input.sourceBytes,
 	);
-	const configured = Number(
-		process.env.MEDIA_PROCESSING_DAILY_BUDGET_GIB ?? "512",
-	);
-	if (
-		!Number.isFinite(configured) ||
-		configured <= 0 ||
-		!Number.isSafeInteger(configured * GiB)
-	)
-		throw new Error("Invalid daily processing budget");
 	const attemptKey = key([
 		"attempt",
 		input.videoId,
@@ -54,13 +44,11 @@ export async function reserveMediaProcessingBudget(input: {
 		input.attemptId,
 	]);
 	const recordingKey = key(["recording", input.videoId, input.generation]);
-	const dailyKey = key(["daily", now.toISOString().slice(0, 10)]);
 	const expiresAt = new Date(now.getTime() + 30 * 24 * 60 * 60_000);
 	return db().transaction(async (tx) => {
 		const limits = [
 			{ id: attemptKey, limitBytes: attemptBytes },
 			{ id: recordingKey, limitBytes: recordingBytes },
-			{ id: dailyKey, limitBytes: configured * GiB },
 		].sort((left, right) => left.id.localeCompare(right.id));
 		await tx
 			.insert(mediaProcessingBudgets)
@@ -92,10 +80,7 @@ export async function reserveMediaProcessingBudget(input: {
 				throw new Error("Processing reservation changed");
 			return attemptBytes;
 		}
-		for (const [id, scope] of [
-			[recordingKey, "recording"],
-			[dailyKey, "daily"],
-		] as const) {
+		for (const [id, scope] of [[recordingKey, "recording"]] as const) {
 			const row = rows.get(id);
 			if (!row || row.reservedBytes + attemptBytes > row.limitBytes)
 				throw new MediaProcessingBudgetError(scope);
diff --git a/apps/web/workflows/finalize-desktop-recording.ts b/apps/web/workflows/finalize-desktop-recording.ts
index 94d086f4a96..90c6f3c7ccb 100644
--- a/apps/web/workflows/finalize-desktop-recording.ts
+++ b/apps/web/workflows/finalize-desktop-recording.ts
@@ -13,7 +13,6 @@ import {
 	claimProcessingAttempt,
 	type DesktopRecordingAttempt,
 	type DesktopRecordingAttemptFence,
-	deferForDailyProcessingBudget,
 	deferWithoutMediaServer,
 	ensureSegmentProcessingJob,
 	getProcessingState,
@@ -140,7 +139,14 @@ export async function finalizeDesktopRecordingWorkflow(
 				break;
 			}
 			const started = await startDesktopRecordingJob(attempt);
-			if (typeof started === "object") continue;
+			if (typeof started === "object") {
+				if (started.status === "capacity") {
+					retainedAttempt = attempt;
+					await sleep(COMPLETION_POLL_INTERVAL_MS);
+				}
+				continue;
+			}
+			const deadline = new Date(Date.now() + ATTEMPT_MAX_DURATION_MS);
 			const jobId = started;
 			for (;;) {
 				await sleep(COMPLETION_POLL_INTERVAL_MS);
@@ -149,9 +155,7 @@ export async function finalizeDesktopRecordingWorkflow(
 					generation: attempt.generation,
 					attemptId: attempt.attemptId,
 					jobId,
-					deadline: new Date(
-						attempt.updatedAt.getTime() + ATTEMPT_MAX_DURATION_MS,
-					),
+					deadline,
 				});
 				if (status === "verified") {
 					completedJobId = jobId;
@@ -409,7 +413,7 @@ async function buildDesktopSegmentsOutput({
 
 export async function startDesktopRecordingJob(
 	attempt: DesktopRecordingAttempt,
-): Promise<string | undefined | { status: "deferred" }> {
+): Promise<string | undefined | { status: "deferred" | "capacity" }> {
 	"use step";
 
 	const current = await getProcessingState(attempt);
@@ -456,10 +460,6 @@ export async function startDesktopRecordingJob(
 		});
 	} catch (error) {
 		if (error instanceof MediaProcessingBudgetError) {
-			if (error.scope === "daily") {
-				await deferForDailyProcessingBudget(attempt);
-				return { status: "deferred" };
-			}
 			await markSourceBlocked({
 				videoId: current.videoId,
 				generation: current.generation,
@@ -540,6 +540,18 @@ export async function startDesktopRecordingJob(
 				signal: AbortSignal.timeout(30_000),
 			},
 		);
+		if (response.status === 503) {
+			const result: unknown = await response.json();
+			if (
+				result &&
+				typeof result === "object" &&
+				"code" in result &&
+				result.code === "SERVER_BUSY" &&
+				(await heartbeatAttempt(attempt))
+			) {
+				return { status: "capacity" };
+			}
+		}
 		if (response.ok) {
 			const result = z
 				.object({

From 382fe3e9113344827b5793688e0fdc5acb369d2a Mon Sep 17 00:00:00 2001
From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com>
Date: Sun, 6 Sep 2026 20:37:27 +0100
Subject: [PATCH 4/6] fix: persist recording capacity waits and honor worker
 backoff

---
 .../unit/desktop-recording-jobs.test.ts       | 53 ++++++++++++++++
 .../unit/finalize-desktop-recording.test.ts   | 28 ++++++++-
 .../unit/media-server-backpressure.test.ts    | 17 ++++++
 apps/web/lib/desktop-recording-jobs.ts        | 60 +++++++++++++++++++
 apps/web/lib/media-server-backpressure.ts     | 28 +++++++--
 .../workflows/finalize-desktop-recording.ts   | 26 ++++++--
 6 files changed, 202 insertions(+), 10 deletions(-)

diff --git a/apps/web/__tests__/unit/desktop-recording-jobs.test.ts b/apps/web/__tests__/unit/desktop-recording-jobs.test.ts
index 5a87cf19ae7..ea096e418fb 100644
--- a/apps/web/__tests__/unit/desktop-recording-jobs.test.ts
+++ b/apps/web/__tests__/unit/desktop-recording-jobs.test.ts
@@ -16,6 +16,7 @@ import {
 	persistSourceCommitCheckpoint,
 	retireDesktopRecordingJobForOutputReplacement,
 	scheduleRetry,
+	waitForDesktopRecordingCapacity,
 } from "@/lib/desktop-recording-jobs";
 import type {
 	RecordingUploadReceipt,
@@ -600,6 +601,58 @@ describe("late verification and source commitment", () => {
 });
 
 describe("retained-source retry policy", () => {
+	it("persists capacity waiting while retaining the attempt and extending its lease past backoff", async () => {
+		const attempt = await createAttempt();
+		Object.assign(getJobRow(), {
+			state: "processing",
+			source,
+			remoteJobId: null,
+			attemptCount: 5,
+		});
+		expect(
+			await waitForDesktopRecordingCapacity({
+				...attempt,
+				now,
+				retryAfterMs: 320_000,
+			}),
+		).toBe(true);
+		expect(getJobRow()).toMatchObject({
+			state: "processing",
+			source,
+			attemptId: attempt.attemptId,
+			attemptCount: 5,
+			output: { kind: "desktop-recording-capacity-wait" },
+			nextRetryAt: new Date(now.getTime() + 320_000),
+		});
+		expect((getJobRow().leaseExpiresAt as Date).getTime()).toBeGreaterThan(
+			now.getTime() + 320_000,
+		);
+		expect(rows.uploads?.[0]?.processingMessage).toContain(
+			"Waiting for a processing slot",
+		);
+	});
+
+	it.each(["owned", "expired"])(
+		"does not overwrite %s work with a capacity wait",
+		async (condition) => {
+			const attempt = await createAttempt();
+			Object.assign(getJobRow(), {
+				state: "processing",
+				source,
+				...(condition === "owned"
+					? { remoteJobId: "worker" }
+					: { leaseExpiresAt: now }),
+			});
+			expect(
+				await waitForDesktopRecordingCapacity({
+					...attempt,
+					now,
+					retryAfterMs: 30_000,
+				}),
+			).toBe(false);
+		},
+	);
+
 	it.each([null, source])(
 		"does not recreate a recording while deletion is pending",
 		async (retainedSource) => {
diff --git a/apps/web/__tests__/unit/finalize-desktop-recording.test.ts b/apps/web/__tests__/unit/finalize-desktop-recording.test.ts
index 0e6c9aafb08..69fd584574e 100644
--- a/apps/web/__tests__/unit/finalize-desktop-recording.test.ts
+++ b/apps/web/__tests__/unit/finalize-desktop-recording.test.ts
@@ -13,6 +13,7 @@ const mocks = vi.hoisted(() => ({
 	ensure: vi.fn(),
 	persist: vi.fn(),
 	heartbeat: vi.fn(),
+	waitCapacity: vi.fn(),
 	blocked: vi.fn(),
 	retry: vi.fn(),
 	attach: vi.fn(),
@@ -59,6 +60,7 @@ vi.mock("@/lib/desktop-recording-jobs", () => ({
 	initializeSourceCommitCheckpoint: mocks.checkpoint,
 	persistSourceCommitCheckpoint: mocks.saveCheckpoint,
 	heartbeatAttempt: mocks.heartbeat,
+	waitForDesktopRecordingCapacity: mocks.waitCapacity,
 	markSourceBlocked: mocks.blocked,
 	scheduleRetry: mocks.retry,
 	attachRemoteJob: mocks.attach,
@@ -179,6 +181,7 @@ beforeEach(() => {
 		withCurrent({ source: savedSource, state: "processing" });
 		return true;
 	});
+	mocks.waitCapacity.mockResolvedValue(true);
 	mocks.heartbeat.mockImplementation(async () => {
 		withCurrent({ leaseExpiresAt: new Date(Date.now() + 5 * 60_000) });
 		return true;
@@ -517,7 +520,10 @@ describe("source commitment and media request compatibility", () => {
 		withCurrent({ state: "retry", leaseExpiresAt: null, attemptCount: 4 });
 		for (let index = 0; index < 8; index++)
 			mocks.fetch.mockResolvedValueOnce(
-				Response.json({ code: "SERVER_BUSY" }, { status: 503 }),
+				Response.json(
+					{ code: "SERVER_BUSY" },
+					{ status: 503, headers: { "Retry-After": "60" } },
+				),
 			);
 		await expect(
 			finalizeDesktopRecordingWorkflow({
@@ -528,6 +534,14 @@ describe("source commitment and media request compatibility", () => {
 		).resolves.toMatchObject({ success: true });
 		expect(current?.attemptCount).toBe(5);
 		expect(mocks.fetch).toHaveBeenCalledTimes(9);
+		expect(mocks.waitCapacity).toHaveBeenCalledWith(
+			expect.objectContaining({ retryAfterMs: expect.any(Number) }),
+		);
+		expect(
+			mocks.sleep.mock.calls.filter(
+				([delay]) => typeof delay === "number" && delay >= 60_000,
+			),
+		).toHaveLength(8);
 		const attempts = mocks.reserveBudget.mock.calls.map(
 			([input]) => input.attemptId,
 		);
@@ -536,6 +550,18 @@ describe("source commitment and media request compatibility", () => {
 		expect(mocks.blocked).not.toHaveBeenCalled();
 	});
 
+	it("respects a persisted capacity wait when a dispatch step is replayed", async () => {
+		withCurrent({
+			output: { kind: "desktop-recording-capacity-wait" },
+			nextRetryAt: new Date(Date.now() + 60_000),
+		});
+		await expect(startDesktopRecordingJob(fixture)).resolves.toMatchObject({
+			status: "capacity",
+		});
+		expect(mocks.fetch).not.toHaveBeenCalled();
+		expect(mocks.sourceUrls).not.toHaveBeenCalled();
+	});
+
 	it("does not treat an ambiguous dispatch failure as a capacity refusal", async () => {
 		mocks.fetch.mockResolvedValueOnce(
 			Response.json({ code: "UPSTREAM_FAILURE" }, { status: 503 }),
diff --git a/apps/web/__tests__/unit/media-server-backpressure.test.ts b/apps/web/__tests__/unit/media-server-backpressure.test.ts
index 039907cbd88..a96e2597481 100644
--- a/apps/web/__tests__/unit/media-server-backpressure.test.ts
+++ b/apps/web/__tests__/unit/media-server-backpressure.test.ts
@@ -2,6 +2,7 @@ import { afterEach, describe, expect, it, vi } from "vitest";
 import { RetryableError } from "workflow";
 import {
 	createMediaServerCapacityError,
+	getMediaServerCapacityDelay,
 	isMediaServerCapacityError,
 } from "@/lib/media-server-backpressure";
 
@@ -61,3 +62,19 @@ describe("media server backpressure", () => {
 		).toBe(15_000);
 	});
 });
+
+it.each(["Sun, 06 Sep 2026 19:40:00 GMT", "1.0001", "999999", "invalid"])(
+	"bounds and rounds server backoff %s",
+	(header) => {
+		vi.useFakeTimers();
+		vi.setSystemTime(new Date("2026-09-06T19:39:00Z"));
+		const delay = getMediaServerCapacityDelay({
+			response: new Response(null, { headers: { "Retry-After": header } }),
+			videoId: "video",
+		});
+		expect(Number.isSafeInteger(delay)).toBe(true);
+		expect(delay).toBeGreaterThan(0);
+		expect(delay).toBeLessThanOrEqual(320_000);
+		if (header.startsWith("Sun")) expect(delay).toBeGreaterThanOrEqual(60_000);
+	},
+);
diff --git a/apps/web/lib/desktop-recording-jobs.ts b/apps/web/lib/desktop-recording-jobs.ts
index fef2fb35f49..bf279ade63f 100644
--- a/apps/web/lib/desktop-recording-jobs.ts
+++ b/apps/web/lib/desktop-recording-jobs.ts
@@ -677,6 +677,66 @@ export async function heartbeatAttempt({
 	});
 }
 
+export async function waitForDesktopRecordingCapacity({
+	now = new Date(),
+	retryAfterMs,
+	...fence
+}: DesktopRecordingAttemptFence & {
+	now?: Date;
+	retryAfterMs: number;
+}): Promise<boolean> {
+	if (
+		!Number.isSafeInteger(retryAfterMs) ||
+		retryAfterMs <= 0 ||
+		retryAfterMs > 360_000
+	)
+		throw new Error("Invalid capacity retry delay");
+	return db().transaction(async (tx) => {
+		const condition = and(
+			attemptCondition(fence),
+			isNull(videoProcessingJobs.remoteJobId),
+			gt(videoProcessingJobs.leaseExpiresAt, now),
+		);
+		const [row] = await tx
+			.select()
+			.from(videoProcessingJobs)
+			.where(condition)
+			.for("update");
+		if (
+			!row ||
+			getDesktopRecordingWorkerCheckpoint(parseDesktopRecordingJob(row))
+		)
+			return false;
+		const nextRetryAt = new Date(now.getTime() + retryAfterMs);
+		await tx
+			.update(videoProcessingJobs)
+			.set({
+				leaseExpiresAt: new Date(
+					nextRetryAt.getTime() + DESKTOP_RECORDING_LEASE_MS,
+				),
+				nextRetryAt,
+				output: {
+					kind: "desktop-recording-capacity-wait",
+					version: 1,
+					retryAt: nextRetryAt.toISOString(),
+				},
+				updatedAt: now,
+			})
+			.where(condition);
+		await tx
+			.update(videoUploads)
+			.set({
+				phase: "processing",
+				processingMessage:
+					"Waiting for a processing slot. Your recording is safely stored.",
+				processingError: null,
+				updatedAt: now,
+			})
+			.where(eq(videoUploads.videoId, fence.videoId));
+		return true;
+	});
+}
+
 export async function scheduleRetry({
 	errorCode,
 	errorMessage,
diff --git a/apps/web/lib/media-server-backpressure.ts b/apps/web/lib/media-server-backpressure.ts
index 3e06aa6818a..b707a15ce25 100644
--- a/apps/web/lib/media-server-backpressure.ts
+++ b/apps/web/lib/media-server-backpressure.ts
@@ -18,10 +18,30 @@ export function createMediaServerCapacityError({
 	videoId: string;
 	priority?: MediaServerJobPriority;
 }): RetryableError {
-	const retryAfterSeconds = Number(response.headers.get("Retry-After"));
+	return new RetryableError(message, {
+		retryAfter: getMediaServerCapacityDelay({ response, videoId, priority }),
+	});
+}
+
+export function getMediaServerCapacityDelay({
+	response,
+	videoId,
+	priority = "normal",
+}: {
+	response: Response;
+	videoId: string;
+	priority?: MediaServerJobPriority;
+}): number {
+	const header = response.headers.get("Retry-After");
+	const retryAfterSeconds =
+		header && /^\d+(?:\.\d+)?$/.test(header)
+			? Number(header)
+			: header
+				? (Date.parse(header) - Date.now()) / 1000
+				: NaN;
 	const minimumDelayMs =
 		Number.isFinite(retryAfterSeconds) && retryAfterSeconds > 0
-			? Math.min(retryAfterSeconds, 300) * 1000
+			? Math.ceil(Math.min(retryAfterSeconds, 300) * 1000)
 			: 15_000;
 	const stableOffset = Array.from(videoId).reduce(
 		(total, character) => (total * 31 + character.charCodeAt(0)) % 20_000,
@@ -29,7 +49,5 @@ export function createMediaServerCapacityError({
 	);
 	const priorityDelayMs = priority === "bulk" ? 15_000 : 0;
 
-	return new RetryableError(message, {
-		retryAfter: minimumDelayMs + priorityDelayMs + stableOffset,
-	});
+	return minimumDelayMs + priorityDelayMs + stableOffset;
 }
diff --git a/apps/web/workflows/finalize-desktop-recording.ts b/apps/web/workflows/finalize-desktop-recording.ts
index 90c6f3c7ccb..3000dee6bbe 100644
--- a/apps/web/workflows/finalize-desktop-recording.ts
+++ b/apps/web/workflows/finalize-desktop-recording.ts
@@ -22,6 +22,7 @@ import {
 	persistCommittedSource,
 	persistSourceCommitCheckpoint,
 	scheduleRetry,
+	waitForDesktopRecordingCapacity,
 } from "@/lib/desktop-recording-jobs";
 import {
 	advanceDesktopRecordingSourceCommit,
@@ -34,6 +35,7 @@ import {
 	MediaProcessingBudgetError,
 	reserveMediaProcessingBudget,
 } from "@/lib/media-processing-budget";
+import { getMediaServerCapacityDelay } from "@/lib/media-server-backpressure";
 import { transcribeVideo } from "@/lib/transcribe";
 import { decodeStorageVideo } from "@/lib/video-storage";
 import { runWorkflowPromise } from "@/lib/workflow-runtime";
@@ -142,7 +144,7 @@ export async function finalizeDesktopRecordingWorkflow(
 			if (typeof started === "object") {
 				if (started.status === "capacity") {
 					retainedAttempt = attempt;
-					await sleep(COMPLETION_POLL_INTERVAL_MS);
+					await sleep(started.retryAfterMs);
 				}
 				continue;
 			}
@@ -413,7 +415,12 @@ async function buildDesktopSegmentsOutput({
 
 export async function startDesktopRecordingJob(
 	attempt: DesktopRecordingAttempt,
-): Promise<string | undefined | { status: "deferred" | "capacity" }> {
+): Promise<
+	| string
+	| undefined
+	| { status: "deferred" }
+	| { status: "capacity"; retryAfterMs: number }
+> {
 	"use step";
 
 	const current = await getProcessingState(attempt);
@@ -422,6 +429,13 @@ export async function startDesktopRecordingJob(
 	}
 	if (current.remoteJobId) return current.remoteJobId;
 	if (current.state === "retry") return { status: "deferred" };
+	const remainingCapacityWait = current.nextRetryAt.getTime() - Date.now();
+	if (
+		current.output?.kind === "desktop-recording-capacity-wait" &&
+		remainingCapacityWait > 0
+	) {
+		return { status: "capacity", retryAfterMs: remainingCapacityWait };
+	}
 	const [video] = await db()
 		.select()
 		.from(videos)
@@ -542,14 +556,18 @@ export async function startDesktopRecordingJob(
 		);
 		if (response.status === 503) {
 			const result: unknown = await response.json();
+			const retryAfterMs = getMediaServerCapacityDelay({
+				response,
+				videoId: attempt.videoId,
+			});
 			if (
 				result &&
 				typeof result === "object" &&
 				"code" in result &&
 				result.code === "SERVER_BUSY" &&
-				(await heartbeatAttempt(attempt))
+				(await waitForDesktopRecordingCapacity({ ...attempt, retryAfterMs }))
 			) {
-				return { status: "capacity" };
+				return { status: "capacity", retryAfterMs };
 			}
 		}
 		if (response.ok) {

From 05e9bc67fbcae7fc8da19cf9f40f10b4169ed2ca Mon Sep 17 00:00:00 2001
From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com>
Date: Sun, 6 Sep 2026 20:43:01 +0100
Subject: [PATCH 5/6] fix: narrow persisted recording capacity checkpoints

---
 apps/web/workflows/finalize-desktop-recording.ts | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/apps/web/workflows/finalize-desktop-recording.ts b/apps/web/workflows/finalize-desktop-recording.ts
index 3000dee6bbe..c211271deda 100644
--- a/apps/web/workflows/finalize-desktop-recording.ts
+++ b/apps/web/workflows/finalize-desktop-recording.ts
@@ -431,7 +431,10 @@ export async function startDesktopRecordingJob(
 	if (current.state === "retry") return { status: "deferred" };
 	const remainingCapacityWait = current.nextRetryAt.getTime() - Date.now();
 	if (
-		current.output?.kind === "desktop-recording-capacity-wait" &&
+		current.output &&
+		typeof current.output === "object" &&
+		"kind" in current.output &&
+		current.output.kind === "desktop-recording-capacity-wait" &&
 		remainingCapacityWait > 0
 	) {
 		return { status: "capacity", retryAfterMs: remainingCapacityWait };

From 756577bb6d3568f25942e662908d17b5b96116ca Mon Sep 17 00:00:00 2001
From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com>
Date: Sun, 6 Sep 2026 20:43:01 +0100
Subject: [PATCH 6/6] fix: prioritize interrupted recordings over incomplete
 upload backlog

---
 .../unit/desktop-recording-jobs.test.ts       |  53 +++++++++
 apps/web/lib/desktop-recording-jobs.ts        | 109 ++++++++++--------
 2 files changed, 112 insertions(+), 50 deletions(-)

diff --git a/apps/web/__tests__/unit/desktop-recording-jobs.test.ts b/apps/web/__tests__/unit/desktop-recording-jobs.test.ts
index ea096e418fb..b10f6802c5a 100644
--- a/apps/web/__tests__/unit/desktop-recording-jobs.test.ts
+++ b/apps/web/__tests__/unit/desktop-recording-jobs.test.ts
@@ -11,6 +11,7 @@ import {
 	heartbeatAttempt,
 	initializeSourceCommitCheckpoint,
 	isDesktopRecordingJobRecoverable,
+	listRecoverableSegmentJobs,
 	markSourceBlocked,
 	persistCommittedSource,
 	persistSourceCommitCheckpoint,
@@ -45,6 +46,9 @@ vi.mock("@cap/database/schema", () => {
 			"nextRetryAt",
 			"leaseExpiresAt",
 			"remoteJobId",
+			"errorCode",
+			"verification",
+			"output",
 		]),
 	};
 });
@@ -136,6 +140,12 @@ function createClient() {
 					table = value.table;
 					return query;
 				},
+				innerJoin() {
+					return query;
+				},
+				orderBy() {
+					return query;
+				},
 				where(value: Condition) {
 					condition = value;
 					return query;
@@ -883,3 +893,46 @@ describe("retained-source retry policy", () => {
 		);
 	});
 });
+
+describe("recovery admission", () => {
+	it("prioritizes interrupted new recordings over an older missing-source backlog", async () => {
+		await createAttempt();
+		const current = {
+			...getJobRow(),
+			state: "committing",
+			source: null,
+			leaseExpiresAt: new Date(now.getTime() - 1),
+			nextRetryAt: now,
+		};
+		rows.jobs = Array.from({ length: 30 }, (_, index) => ({
+			...current,
+			videoId: `old-${index}`,
+			state: "source-blocked",
+			errorCode: "source-missing",
+			nextRetryAt: new Date(now.getTime() - 24 * 60 * 60_000),
+		}));
+		rows.jobs.push(current);
+		const selected = await listRecoverableSegmentJobs({ now, limit: 3 });
+		expect(selected[0]?.videoId).toBe(videoId);
+		expect(selected).toHaveLength(3);
+	});
+
+	it("excludes exhausted and intentionally retired jobs before applying the recovery limit", async () => {
+		await createAttempt();
+		const current = {
+			...getJobRow(),
+			state: "retry",
+			source: null,
+			leaseExpiresAt: null,
+			nextRetryAt: now,
+		};
+		rows.jobs = [
+			"processing-retry-exhausted",
+			"output-replaced",
+			"video-deleting",
+		].map((errorCode) => ({ ...current, videoId: errorCode, errorCode }));
+		rows.jobs.push(current);
+		const selected = await listRecoverableSegmentJobs({ now, limit: 1 });
+		expect(selected.map((job) => job.videoId)).toEqual([videoId]);
+	});
+});
diff --git a/apps/web/lib/desktop-recording-jobs.ts b/apps/web/lib/desktop-recording-jobs.ts
index bf279ade63f..55a10c2a8e0 100644
--- a/apps/web/lib/desktop-recording-jobs.ts
+++ b/apps/web/lib/desktop-recording-jobs.ts
@@ -948,60 +948,58 @@ export async function listRecoverableSegmentJobs({
 	limit?: number;
 } = {}): Promise<DesktopRecordingJob[]> {
 	const batchSize = Math.max(1, Math.min(limit, 100));
-	const pending = await db()
-		.select(getTableColumns(videoProcessingJobs))
-		.from(videoProcessingJobs)
-		.innerJoin(videos, eq(videos.id, videoProcessingJobs.videoId))
-		.where(
-			and(
-				inArray(videoProcessingJobs.state, [
-					"committing",
-					"queued",
-					"retry",
-					"source-blocked",
-				]),
-				or(
-					ne(videoProcessingJobs.state, "source-blocked"),
-					isNull(videoProcessingJobs.source),
-				),
-				or(
-					isNull(videoProcessingJobs.errorCode),
-					and(
-						ne(
-							videoProcessingJobs.errorCode,
-							DESKTOP_RECORDING_OUTPUT_REPLACED,
+	const candidates = async (
+		states: DesktopRecordingJob["state"][],
+		candidateLimit: number,
+		byLease = false,
+	) =>
+		db()
+			.select(getTableColumns(videoProcessingJobs))
+			.from(videoProcessingJobs)
+			.innerJoin(videos, eq(videos.id, videoProcessingJobs.videoId))
+			.where(
+				and(
+					inArray(videoProcessingJobs.state, states),
+					or(
+						ne(videoProcessingJobs.state, "source-blocked"),
+						isNull(videoProcessingJobs.source),
+					),
+					or(
+						isNull(videoProcessingJobs.errorCode),
+						and(
+							ne(
+								videoProcessingJobs.errorCode,
+								DESKTOP_RECORDING_OUTPUT_REPLACED,
+							),
+							ne(videoProcessingJobs.errorCode, DESKTOP_RECORDING_DELETING),
+							ne(
+								videoProcessingJobs.errorCode,
+								DESKTOP_RECORDING_RETRY_EXHAUSTED,
+							),
 						),
-						ne(videoProcessingJobs.errorCode, DESKTOP_RECORDING_DELETING),
+					),
+					lte(videoProcessingJobs.nextRetryAt, now),
+					or(
+						isNull(videoProcessingJobs.leaseExpiresAt),
+						lte(videoProcessingJobs.leaseExpiresAt, now),
 					),
 				),
-				lte(videoProcessingJobs.nextRetryAt, now),
-				or(
-					isNull(videoProcessingJobs.leaseExpiresAt),
-					lte(videoProcessingJobs.leaseExpiresAt, now),
+			)
+			.orderBy(
+				asc(
+					byLease
+						? videoProcessingJobs.leaseExpiresAt
+						: videoProcessingJobs.nextRetryAt,
 				),
-			),
-		)
-		.orderBy(
-			asc(videoProcessingJobs.nextRetryAt),
-			asc(videoProcessingJobs.videoId),
-		)
-		.limit(batchSize);
-	const expired = await db()
-		.select(getTableColumns(videoProcessingJobs))
-		.from(videoProcessingJobs)
-		.innerJoin(videos, eq(videos.id, videoProcessingJobs.videoId))
-		.where(
-			and(
-				eq(videoProcessingJobs.state, "processing"),
-				lte(videoProcessingJobs.leaseExpiresAt, now),
-			),
-		)
-		.orderBy(
-			asc(videoProcessingJobs.leaseExpiresAt),
-			asc(videoProcessingJobs.videoId),
-		)
-		.limit(batchSize);
-	return [...pending, ...expired]
+				asc(videoProcessingJobs.videoId),
+			)
+			.limit(candidateLimit);
+	const pending = await candidates(
+		["committing", "queued", "retry"],
+		batchSize,
+	);
+	const expired = await candidates(["processing"], batchSize, true);
+	const active = [...pending, ...expired]
 		.map(parseDesktopRecordingJob)
 		.filter((job) => isDesktopRecordingJobRecoverable(job, now))
 		.sort((left, right) => {
@@ -1013,4 +1011,15 @@ export async function listRecoverableSegmentJobs({
 			);
 		})
 		.slice(0, batchSize);
+	if (active.length === batchSize) return active;
+	const blocked = await candidates(
+		["source-blocked"],
+		batchSize - active.length,
+	);
+	return [
+		...active,
+		...blocked
+			.map(parseDesktopRecordingJob)
+			.filter((job) => isDesktopRecordingJobRecoverable(job, now)),
+	];
 }