fix: persist recording capacity waits and honor worker backoff

Author: richiemcilroy

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

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 }),

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

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,

apps/web/lib/media-server-backpressure.ts

@@ -18,18 +18,36 @@ 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,
 		0,
 	);
 	const priorityDelayMs = priority === "bulk" ? 15_000 : 0;
 
-	return new RetryableError(message, {
-		retryAfter: minimumDelayMs + priorityDelayMs + stableOffset,
-	});
+	return minimumDelayMs + priorityDelayMs + stableOffset;
 }

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