fix: prioritize interrupted recordings over incomplete upload backlog

Author: richiemcilroy

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

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