fix: process recordings without a daily transfer allowance

Author: richiemcilroy

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) => {

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,40 +513,35 @@ 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,
 				userId,
 				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 () => {

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");
+});

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,