fix: bound recording retries and avoid wasted Drive downloads

Author: richiemcilroy

apps/web/__tests__/unit/desktop-recording-jobs.test.ts

@@ -720,6 +720,65 @@ describe("retained-source retry policy", () => {
 		});
 	});
 
+	it("pauses a repeatedly failing recording and retains its source", async () => {
+		const attempt = await createAttempt();
+		await persistCommittedSource(attempt, source);
+		Object.assign(rows.jobs[0], { attemptCount: 5 });
+		expect(
+			await scheduleRetry({
+				...attempt,
+				errorCode: "output-invalid",
+				errorMessage: "Timeline mismatch",
+			}),
+		).toBe(true);
+		expect(rows.jobs[0]).toMatchObject({
+			state: "source-blocked",
+			source,
+			errorCode: "processing-retry-exhausted",
+			errorMessage: "output-invalid: Timeline mismatch",
+		});
+		expect(rows.uploads[0]).toMatchObject({ phase: "error" });
+		vi.setSystemTime(new Date(now.getTime() + 24 * 60 * 60_000));
+		expect(
+			await claimProcessingAttempt({ videoId, generation: attempt.generation }),
+		).toBeNull();
+		expect(rows.jobs[0]?.attemptCount).toBe(5);
+	});
+
+	it("pauses an exhausted legacy job before downloading its source again", async () => {
+		const attempt = await createAttempt();
+		Object.assign(rows.jobs[0], {
+			attemptCount: 220,
+			state: "retry",
+			leaseExpiresAt: null,
+			nextRetryAt: now,
+		});
+		expect(
+			await claimProcessingAttempt({ videoId, generation: attempt.generation }),
+		).toBeNull();
+		expect(rows.jobs[0]).toMatchObject({
+			attemptCount: 220,
+			errorCode: "processing-retry-exhausted",
+		});
+		await ensureSegmentProcessingJob({ videoId, userId });
+		expect(rows.jobs[0]).toMatchObject({
+			state: "source-blocked",
+			errorCode: "processing-retry-exhausted",
+		});
+	});
+
+	it("does not interrupt an active final attempt", async () => {
+		const attempt = await createAttempt();
+		Object.assign(rows.jobs[0], { attemptCount: 5 });
+		expect(
+			await claimProcessingAttempt({ videoId, generation: attempt.generation }),
+		).toBeNull();
+		expect(rows.jobs[0]).toMatchObject({
+			state: "committing",
+			attemptId: attempt.attemptId,
+		});
+	});
+
 	it("recovers old jobs regardless of recording age or previous attempt count", async () => {
 		const attempt = await createAttempt();
 		const job: DesktopRecordingJob = {

apps/web/__tests__/unit/storage-object-verification.test.ts

@@ -82,6 +82,33 @@ describe("recording verification object reads", () => {
 		expect(mocks.head).not.toHaveBeenCalled();
 	});
 
+	it("cancels ordinary upstream downloads when the client disconnects", async () => {
+		const controller = new AbortController();
+		const req = new NextRequest(
+			"https://cap.test/api/storage/object?videoId=video&key=owner/video/result.mp4&token=test",
+			{ signal: controller.signal },
+		);
+		await GET(req);
+		expect(mocks.read).toHaveBeenCalledWith("owner/video/result.mp4", null, {
+			signal: req.signal,
+		});
+		controller.abort();
+		expect(mocks.read.mock.calls.at(-1)?.[2].signal.aborted).toBe(true);
+	});
+
+	it("answers ordinary HEAD requests without downloading the video", async () => {
+		const response = await HEAD(
+			new NextRequest(
+				"https://cap.test/api/storage/object?videoId=video&key=owner/video/result.mp4&token=test",
+				{ method: "HEAD" },
+			),
+		);
+		expect(response.status).toBe(200);
+		expect(response.headers.get("Content-Length")).toBe("100");
+		expect(await response.text()).toBe("");
+		expect(mocks.read).not.toHaveBeenCalled();
+	});
+
 	it("serves content-bound HEAD without downloading media", async () => {
 		const identity = `"cap-drive-content-v1:${"a".repeat(64)}"`;
 		mocks.head.mockReturnValue(
@@ -261,7 +288,9 @@ describe("recording verification object reads", () => {
 			mocks.video.source = { type: "desktopMP4", [field]: key };
 			const response = await request({}, key);
 			expect(response.status).toBe(206);
-			expect(mocks.read).toHaveBeenCalledWith(key, null);
+			expect(mocks.read).toHaveBeenCalledWith(key, null, {
+				signal: expect.any(AbortSignal),
+			});
 		},
 	);
 });

apps/web/app/api/storage/object/route.ts

@@ -168,12 +168,14 @@ export async function GET(request: NextRequest) {
 		const verificationRequested =
 			request.headers.get("x-cap-recording-verification") === "1";
 		const expectedIdentity = request.headers.get("if-match");
-		const head = verificationRequested
-			? yield* storage.headObject(key)
-			: undefined;
-		const identity = head
-			? getRecordingObjectIdentity(head, expectedIdentity ?? undefined)
-			: undefined;
+		const head =
+			verificationRequested || request.method === "HEAD"
+				? yield* storage.headObject(key)
+				: undefined;
+		const identity =
+			verificationRequested && head
+				? getRecordingObjectIdentity(head, expectedIdentity ?? undefined)
+				: undefined;
 		if (verificationRequested && !identity) {
 			return new Response("Object identity is unavailable", { status: 503 });
 		}
@@ -184,14 +186,9 @@ export async function GET(request: NextRequest) {
 		) {
 			return new Response("Object changed", { status: 412 });
 		}
-		if (
-			verificationRequested &&
-			request.method === "HEAD" &&
-			head &&
-			identity
-		) {
+		if (request.method === "HEAD" && head) {
 			const headers = new Headers(CACHE_CONTROL_HEADERS);
-			headers.set("ETag", identity);
+			if (identity) headers.set("ETag", identity);
 			headers.set("Accept-Ranges", "bytes");
 			if (head.ContentLength !== undefined)
 				headers.set("Content-Length", String(head.ContentLength));
@@ -203,7 +200,9 @@ export async function GET(request: NextRequest) {
 					objectIdentity: identity,
 					signal: request.signal,
 				})
-			: yield* storage.getObjectResponse(key, request.headers.get("range"));
+			: yield* storage.getObjectResponse(key, request.headers.get("range"), {
+					signal: request.signal,
+				});
 		const headers = new Headers(CACHE_CONTROL_HEADERS);
 		if (identity) headers.set("ETag", identity);
 		copyHeader(upstream.headers, headers, "content-type", "Content-Type");

apps/web/lib/desktop-recording-jobs.ts

@@ -33,6 +33,10 @@ export const DESKTOP_RECORDING_LEASE_MS = 5 * 60 * 1_000;
 export const DESKTOP_RECORDING_SOURCE_RETRY_MS = 60 * 60 * 1_000;
 export const DESKTOP_RECORDING_OUTPUT_REPLACED = "output-replaced";
 export const DESKTOP_RECORDING_DELETING = "video-deleting";
+export const DESKTOP_RECORDING_RETRY_EXHAUSTED = "processing-retry-exhausted";
+const MAX_AUTOMATIC_ATTEMPTS = 5;
+const RETRY_EXHAUSTED_MESSAGE =
+	"Processing paused after repeated failures. Your uploaded recording is retained. Please contact support.";
 
 const sourceSchema = z.object({
 	version: z.literal(1),
@@ -158,6 +162,7 @@ export function isDesktopRecordingJobRecoverable(
 	now: Date,
 ) {
 	if (job.state === "verified") return false;
+	if (job.errorCode === DESKTOP_RECORDING_RETRY_EXHAUSTED) return false;
 	if (job.errorCode === DESKTOP_RECORDING_OUTPUT_REPLACED) return false;
 	if (job.errorCode === DESKTOP_RECORDING_DELETING) return false;
 	if (job.state === "source-blocked" && job.source) return false;
@@ -333,7 +338,11 @@ export async function ensureSegmentProcessingJob({
 					.where(eq(videoProcessingJobs.videoId, videoId));
 			}
 		}
-		if (job.state === "source-blocked" && !job.source) {
+		if (
+			job.state === "source-blocked" &&
+			!job.source &&
+			job.errorCode !== DESKTOP_RECORDING_RETRY_EXHAUSTED
+		) {
 			job = {
 				...job,
 				state: "committing",
@@ -374,6 +383,33 @@ export async function getProcessingState({
 	return row ? parseDesktopRecordingJob(row) : null;
 }
 
+async function pauseExhaustedRecording(
+	tx: Parameters<Parameters<ReturnType<typeof db>["transaction"]>[0]>[0],
+	videoId: Video.VideoId,
+	errorMessage: string,
+	now: Date,
+) {
+	await tx
+		.update(videoProcessingJobs)
+		.set({
+			state: "source-blocked",
+			leaseExpiresAt: null,
+			errorCode: DESKTOP_RECORDING_RETRY_EXHAUSTED,
+			errorMessage,
+			updatedAt: now,
+		})
+		.where(eq(videoProcessingJobs.videoId, videoId));
+	await tx
+		.update(videoUploads)
+		.set({
+			phase: "error",
+			processingMessage: RETRY_EXHAUSTED_MESSAGE,
+			processingError: RETRY_EXHAUSTED_MESSAGE,
+			updatedAt: now,
+		})
+		.where(eq(videoUploads.videoId, videoId));
+}
+
 export async function claimProcessingAttempt({
 	videoId,
 	generation,
@@ -397,6 +433,15 @@ export async function claimProcessingAttempt({
 		if (!row) return null;
 		const job = parseDesktopRecordingJob(row);
 		if (!isDesktopRecordingJobRecoverable(job, now)) return null;
+		if (job.attemptCount >= MAX_AUTOMATIC_ATTEMPTS) {
+			await pauseExhaustedRecording(
+				tx,
+				videoId,
+				job.errorMessage ?? "Automatic processing attempt limit reached.",
+				now,
+			);
+			return null;
+		}
 		const attempt: DesktopRecordingAttempt = {
 			...job,
 			state: job.source ? "processing" : "committing",
@@ -651,6 +696,15 @@ export async function scheduleRetry({
 			.where(attemptCondition(fence))
 			.for("update");
 		if (!row) return false;
+		if (row.attemptCount >= MAX_AUTOMATIC_ATTEMPTS) {
+			await pauseExhaustedRecording(
+				tx,
+				fence.videoId,
+				`${errorCode}: ${errorMessage}`,
+				now,
+			);
+			return true;
+		}
 		await tx
 			.update(videoProcessingJobs)
 			.set({

packages/web-backend/src/Storage/GoogleDrive.ts

@@ -1373,6 +1373,7 @@ export const getGoogleDriveObjectResponse = (
 	fileId: string,
 	range?: string | null,
 	tokenStore?: GoogleDriveTokenStore,
+	signal?: AbortSignal,
 ) =>
 	Effect.gen(function* () {
 		const headers: Record<string, string> = {};
@@ -1382,7 +1383,7 @@ export const getGoogleDriveObjectResponse = (
 			appendSharedDriveCreateParams(
 				`${DRIVE_API_BASE}/files/${encodeURIComponent(fileId)}?alt=media`,
 			),
-			{ headers },
+			{ headers, signal },
 			tokenStore,
 		);
 

packages/web-backend/src/Storage/index.ts

@@ -1033,11 +1033,11 @@ const makeGoogleDriveAccess = ({
 		getObjectResponse: (
 			key: string,
 			range?: string | null,
-			verification?: GoogleDriveRecordingRead,
+			verification?: GoogleDriveRecordingRead | { signal: AbortSignal },
 		) =>
 			getObjectRecord(key).pipe(
 				Effect.flatMap((object) =>
-					verification
+					verification && "objectIdentity" in verification
 						? getGoogleDriveRecordingResponse(
 								config,
 								object.providerObjectId,
@@ -1046,7 +1046,13 @@ const makeGoogleDriveAccess = ({
 								tokenStore,
 							)
 						: withRecoveredDriveFile(key, object, (fileId) =>
-								getGoogleDriveObjectResponse(config, fileId, range, tokenStore),
+								getGoogleDriveObjectResponse(
+									config,
+									fileId,
+									range,
+									tokenStore,
+									verification?.signal,
+								),
 							),
 				),
 			),