fix: restrict automatic recovery to actionable recording sources

Author: richiemcilroy

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

@@ -895,6 +895,29 @@ describe("retained-source retry policy", () => {
 });
 
 describe("recovery admission", () => {
+	it("keeps an inspected missing source paused until a completion request resumes it", async () => {
+		const attempt = await createAttempt();
+		const paused = {
+			...attempt,
+			state: "source-blocked" as const,
+			source: null,
+			leaseExpiresAt: null,
+			errorCode: "source-reupload-required",
+			nextRetryAt: now,
+		};
+		rows.jobs = [paused];
+		expect(isDesktopRecordingJobRecoverable(paused, now)).toBe(false);
+		expect(await listRecoverableSegmentJobs({ now })).toEqual([]);
+		const resumed = await ensureSegmentProcessingJob({
+			videoId,
+			userId,
+			verification,
+			now,
+		});
+		expect(resumed.job).toMatchObject({ state: "committing", errorCode: null });
+		expect(isDesktopRecordingJobRecoverable(resumed.job, now)).toBe(true);
+	});
+
 	it("prioritizes interrupted new recordings over an older missing-source backlog", async () => {
 		await createAttempt();
 		const current = {
@@ -928,6 +951,7 @@ describe("recovery admission", () => {
 		};
 		rows.jobs = [
 			"processing-retry-exhausted",
+			"source-reupload-required",
 			"output-replaced",
 			"video-deleting",
 		].map((errorCode) => ({ ...current, videoId: errorCode, errorCode }));

apps/web/__tests__/unit/desktop-segments-recovery.test.ts

@@ -16,6 +16,7 @@ vi.mock("@cap/database/schema", () => ({
 		videoId: "upload.videoId",
 		phase: "upload.phase",
 		updatedAt: "upload.updatedAt",
+		startedAt: "upload.startedAt",
 	},
 	videoProcessingJobs: { videoId: "job.videoId" },
 }));
@@ -26,6 +27,7 @@ vi.mock("drizzle-orm", () => ({
 	inArray: (left: unknown, right: unknown) => ({ in: [left, right] }),
 	isNull: (value: unknown) => ({ null: value }),
 	lte: (left: unknown, right: unknown) => ({ lte: [left, right] }),
+	gte: (left: unknown, right: unknown) => ({ gte: [left, right] }),
 	sql: (strings: TemplateStringsArray, ...values: unknown[]) => ({
 		strings: [...strings],
 		values,
@@ -159,6 +161,38 @@ describe("committed source recovery", () => {
 });
 
 describe("durable recovery scheduling", () => {
+	it("does not create a durable job for an unfinished legacy upload", async () => {
+		const now = new Date("2026-09-09T20:00:00Z");
+		const legacy = selectChain([{ videoId, ownerId: userId }]);
+		mocks.db.mockReturnValueOnce(legacy).mockReturnValue(selectChain([video]));
+		mocks.get.mockReturnValue(
+			Effect.succeed(
+				Option.some(JSON.stringify({ ...manifest, is_complete: false })),
+			),
+		);
+		const result = await recoverStaleDesktopSegments({ now });
+		expect(result.statuses).toEqual({ "source-incomplete": 1 });
+		expect(mocks.queue).not.toHaveBeenCalled();
+		expect(mocks.put).not.toHaveBeenCalled();
+		expect(JSON.stringify(legacy.where.mock.calls)).toContain(
+			"2026-09-02T20:00:00.000Z",
+		);
+	});
+
+	it("continues inspecting legacy uploads after a storage failure", async () => {
+		const legacy = selectChain([
+			{ videoId, ownerId: userId },
+			{ videoId: "second", ownerId: userId },
+		]);
+		mocks.db.mockReturnValueOnce(legacy).mockReturnValue(selectChain([video]));
+		mocks.get.mockReturnValueOnce(
+			Effect.fail(new Error("storage unavailable")),
+		);
+		const result = await recoverStaleDesktopSegments();
+		expect(result.statuses).toEqual({ failed: 1, queued: 1 });
+		expect(mocks.queue).toHaveBeenCalledTimes(1);
+	});
+
 	it("recovers old retry and expired processing jobs without a recording-age cutoff", async () => {
 		mocks.db.mockReturnValue(selectChain([]));
 		mocks.recoverable.mockResolvedValue([
@@ -192,14 +226,14 @@ describe("durable recovery scheduling", () => {
 
 	it("adopts stranded legacy processing rows without assuming their inventory is complete", async () => {
 		const chain = selectChain([{ videoId, ownerId: userId }]);
-		mocks.db.mockReturnValue(chain);
+		mocks.db.mockReturnValueOnce(chain).mockReturnValue(selectChain([video]));
 		mocks.queue.mockRejectedValue(new SourceCommitPendingError());
 		const result = await recoverStaleDesktopSegments();
 		expect(result.statuses).toEqual({ "source-committing": 1 });
 		const query = JSON.stringify(chain.where.mock.calls);
 		expect(query).toContain('"processing"');
 		expect(query).not.toContain("28 HOUR");
-		expect(query).not.toContain("startedAt");
+		expect(query).toContain("startedAt");
 		expect(mocks.put).not.toHaveBeenCalled();
 	});
 
@@ -231,7 +265,7 @@ describe("durable recovery scheduling", () => {
 				ownerId: userId,
 			})),
 		);
-		mocks.db.mockReturnValue(legacy);
+		mocks.db.mockReturnValueOnce(legacy).mockReturnValue(selectChain([video]));
 		const result = await recoverStaleDesktopSegments();
 		expect(mocks.recoverable).toHaveBeenCalledWith(
 			expect.objectContaining({ limit: 15 }),

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

@@ -33,6 +33,8 @@ 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_SOURCE_REUPLOAD_REQUIRED =
+	"source-reupload-required";
 export const DESKTOP_RECORDING_RETRY_EXHAUSTED = "processing-retry-exhausted";
 const MAX_AUTOMATIC_ATTEMPTS = 5;
 const RETRY_EXHAUSTED_MESSAGE =
@@ -162,6 +164,8 @@ export function isDesktopRecordingJobRecoverable(
 	now: Date,
 ) {
 	if (job.state === "verified") return false;
+	if (job.errorCode === DESKTOP_RECORDING_SOURCE_REUPLOAD_REQUIRED)
+		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;
@@ -972,6 +976,10 @@ export async function listRecoverableSegmentJobs({
 								DESKTOP_RECORDING_OUTPUT_REPLACED,
 							),
 							ne(videoProcessingJobs.errorCode, DESKTOP_RECORDING_DELETING),
+							ne(
+								videoProcessingJobs.errorCode,
+								DESKTOP_RECORDING_SOURCE_REUPLOAD_REQUIRED,
+							),
 							ne(
 								videoProcessingJobs.errorCode,
 								DESKTOP_RECORDING_RETRY_EXHAUSTED,

apps/web/lib/desktop-segments-recovery.ts

@@ -6,7 +6,7 @@ import {
 } from "@cap/database/schema";
 import { Storage } from "@cap/web-backend";
 import { type User, Video } from "@cap/web-domain";
-import { and, asc, eq, inArray, isNull, lte, sql } from "drizzle-orm";
+import { and, asc, eq, gte, inArray, isNull, lte, sql } from "drizzle-orm";
 import { Effect, Option, Schema } from "effect";
 import {
 	DesktopRecordingSourceBlockedError,
@@ -23,6 +23,8 @@ import { decodeStorageVideo } from "@/lib/video-storage";
 
 export const DESKTOP_SEGMENTS_RECOVERY_MIN_AGE_MS = 60 * 60 * 1_000;
 export const DESKTOP_SEGMENTS_RECOVERY_BATCH_SIZE = 20;
+export const DESKTOP_SEGMENTS_LEGACY_RECOVERY_MAX_AGE_MS =
+	7 * 24 * 60 * 60 * 1_000;
 
 const RECOVERABLE_UPLOAD_PHASES = [
 	"uploading",
@@ -244,20 +246,33 @@ export async function recoverStaleDesktopSegments({
 			and(
 				inArray(videoUploads.phase, RECOVERABLE_UPLOAD_PHASES),
 				lte(videoUploads.updatedAt, staleBefore),
+				gte(
+					videoUploads.startedAt,
+					new Date(now.getTime() - DESKTOP_SEGMENTS_LEGACY_RECOVERY_MAX_AGE_MS),
+				),
 				isNull(videoProcessingJobs.videoId),
 				sql`JSON_UNQUOTE(JSON_EXTRACT(${videos.source}, '$.type')) = 'desktopSegments'`,
 			),
 		)
 		.orderBy(asc(videoUploads.updatedAt), asc(videoUploads.videoId))
 		.limit(remaining);
 	for (const candidate of legacy) {
-		record(
-			candidate.videoId,
-			await recoverRecording({
+		try {
+			const result = await completeDesktopSegmentsManifestAndQueue({
 				videoId: candidate.videoId,
 				userId: candidate.ownerId,
-			}),
-		);
+			});
+			record(candidate.videoId, result.status);
+		} catch (error) {
+			console.error(
+				"[desktop-segments-recovery] Legacy source inspection failed",
+				{
+					videoId: candidate.videoId,
+					error,
+				},
+			);
+			record(candidate.videoId, "failed");
+		}
 	}
 	return summary;
 }