fix(web): stop empty Instant caps from delaying stuck desktop recording recovery

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

@@ -25,7 +25,8 @@ vi.mock("drizzle-orm", () => ({
 	and: (...args: unknown[]) => args,
 	or: (...args: unknown[]) => ({ or: args }),
 	notLike: (left: unknown, right: unknown) => ({ notLike: [left, right] }),
-	asc: (value: unknown) => value,
+	asc: (value: unknown) => ({ asc: value }),
+	desc: (value: unknown) => ({ desc: value }),
 	eq: (left: unknown, right: unknown) => ({ eq: [left, right] }),
 	inArray: (left: unknown, right: unknown) => ({ in: [left, right] }),
 	isNull: (value: unknown) => ({ null: value }),
@@ -190,7 +191,10 @@ 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.db
+			.mockReturnValueOnce(legacy)
+			.mockReturnValueOnce(legacy)
+			.mockReturnValue(selectChain([video]));
 		mocks.get.mockReturnValue(
 			Effect.succeed(
 				Option.some(JSON.stringify({ ...manifest, is_complete: false })),
@@ -210,7 +214,10 @@ describe("durable recovery scheduling", () => {
 			{ videoId, ownerId: userId },
 			{ videoId: "second", ownerId: userId },
 		]);
-		mocks.db.mockReturnValueOnce(legacy).mockReturnValue(selectChain([video]));
+		mocks.db
+			.mockReturnValueOnce(legacy)
+			.mockReturnValueOnce(legacy)
+			.mockReturnValue(selectChain([video]));
 		mocks.get.mockReturnValueOnce(
 			Effect.fail(new Error("storage unavailable")),
 		);
@@ -252,7 +259,10 @@ 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.mockReturnValueOnce(chain).mockReturnValue(selectChain([video]));
+		mocks.db
+			.mockReturnValueOnce(chain)
+			.mockReturnValueOnce(chain)
+			.mockReturnValue(selectChain([video]));
 		mocks.queue.mockRejectedValue(new SourceCommitPendingError());
 		const result = await recoverStaleDesktopSegments();
 		expect(result.statuses).toEqual({ "source-committing": 1 });
@@ -292,7 +302,10 @@ describe("durable recovery scheduling", () => {
 				ownerId: userId,
 			})),
 		);
-		mocks.db.mockReturnValueOnce(legacy).mockReturnValue(selectChain([video]));
+		mocks.db
+			.mockReturnValueOnce(legacy)
+			.mockReturnValueOnce(legacy)
+			.mockReturnValue(selectChain([video]));
 		const result = await recoverStaleDesktopSegments();
 		expect(mocks.recoverable).toHaveBeenCalledWith(
 			expect.objectContaining({ limit: 15 }),
@@ -304,3 +317,197 @@ describe("durable recovery scheduling", () => {
 		).toHaveLength(5);
 	});
 });
+
+describe("legacy recovery order", () => {
+	it("keeps every oldest-first slot and checks the newest eligible rows on top", async () => {
+		const oldest = selectChain([
+			{ videoId: "oldest", ownerId: userId },
+			{ videoId: "both", ownerId: userId },
+		]);
+		const newest = selectChain([
+			{ videoId: "newest", ownerId: userId },
+			{ videoId: "both", ownerId: userId },
+		]);
+		mocks.db
+			.mockReturnValueOnce(oldest)
+			.mockReturnValueOnce(newest)
+			.mockReturnValue(selectChain([video]));
+		mocks.get.mockReturnValue(Effect.succeed(Option.none()));
+		const result = await recoverStaleDesktopSegments();
+		expect(result.results.map((entry) => entry.videoId)).toEqual([
+			"oldest",
+			"both",
+			"newest",
+		]);
+		expect(result.statuses).toEqual({ "missing-manifest": 3 });
+		expect(oldest.orderBy).toHaveBeenCalledWith(
+			{ asc: "upload.updatedAt" },
+			{ asc: "upload.videoId" },
+		);
+		expect(oldest.limit).toHaveBeenCalledWith(20);
+		expect(newest.orderBy).toHaveBeenCalledWith(
+			{ desc: "upload.updatedAt" },
+			{ desc: "upload.videoId" },
+		);
+		expect(newest.limit).toHaveBeenCalledWith(5);
+	});
+
+	type Row = { videoId: string; startedAt: number };
+	const minute = 60_000;
+	const day = 24 * 60 * minute;
+
+	/// Runs the cron every 15 minutes over `rows`, which never change, the way
+	/// an empty cap's upload row never does. Rows whose manifest is finished
+	/// are queued once checked and then leave the legacy pass.
+	async function simulate({
+		rows,
+		finished,
+		from,
+		to,
+		durableJobs,
+	}: {
+		rows: Row[];
+		finished: Set<string>;
+		from: number;
+		to: number;
+		durableJobs: number;
+	}) {
+		const queued = new Set<string>();
+		let now = from;
+		const eligible = () =>
+			rows.filter(
+				(row) =>
+					!queued.has(row.videoId) &&
+					row.startedAt <= now - 60 * minute &&
+					row.startedAt >= now - 7 * day,
+			);
+		mocks.recoverable.mockImplementation(async ({ limit }) =>
+			Array.from({ length: Math.min(limit, durableJobs) }, (_, index) => ({
+				videoId: `durable-${index}`,
+				ownerId: userId,
+				verification: null,
+			})),
+		);
+		mocks.db.mockImplementation(() => {
+			const state: { joined: boolean; order?: unknown; where?: unknown } = {
+				joined: false,
+			};
+			const chain = {
+				select: () => chain,
+				from: () => chain,
+				innerJoin: () => {
+					state.joined = true;
+					return chain;
+				},
+				leftJoin: () => chain,
+				where: (where: unknown) => {
+					state.where = where;
+					return chain;
+				},
+				orderBy: (order: unknown) => {
+					state.order = order;
+					return chain;
+				},
+				limit: async (limit: number) => {
+					if (!state.joined) {
+						const id = (state.where as Array<{ eq?: [string, string] }>).find(
+							(part) => part?.eq?.[0] === "video.id",
+						)?.eq?.[1];
+						return [{ ...video, id }];
+					}
+					const descending = "desc" in (state.order as object);
+					return eligible()
+						.sort((a, b) =>
+							descending
+								? b.startedAt - a.startedAt ||
+									b.videoId.localeCompare(a.videoId)
+								: a.startedAt - b.startedAt ||
+									a.videoId.localeCompare(b.videoId),
+						)
+						.slice(0, limit)
+						.map((row) => ({ videoId: row.videoId, ownerId: userId }));
+				},
+			};
+			return chain;
+		});
+		mocks.get.mockImplementation((key: string) => {
+			const id = key.split("/")[1] ?? "";
+			return Effect.succeed(
+				finished.has(id)
+					? Option.some(JSON.stringify(manifest))
+					: Option.none(),
+			);
+		});
+		mocks.queue.mockImplementation(async ({ videoId: id }) => {
+			if (!id.startsWith("durable-")) queued.add(id);
+			return "queued";
+		});
+		const checkedAt = new Map<string, number>();
+		for (; now <= to; now += 15 * minute) {
+			const result = await recoverStaleDesktopSegments({ now: new Date(now) });
+			for (const entry of result.results) {
+				if (!checkedAt.has(entry.videoId)) checkedAt.set(entry.videoId, now);
+			}
+		}
+		return { queued, checkedAt };
+	}
+
+	function emptyCaps(from: number, to: number) {
+		const rows: Row[] = [];
+		for (let at = from; at < to; at += 15 * minute) {
+			rows.push({ videoId: `empty-${at}`, startedAt: at });
+		}
+		return rows;
+	}
+
+	it("reaches a stuck recording behind a week of empty caps soon after it qualifies, and still checks every row before it ages out", async () => {
+		const start = new Date("2026-10-01T00:00:00Z").getTime();
+		const end = start + day;
+		const rows = emptyCaps(start - 8 * day, end);
+		const stuck = { videoId: "stuck", startedAt: start + 3 * 60 * minute };
+		rows.push(stuck);
+		const { queued, checkedAt } = await simulate({
+			rows,
+			finished: new Set(["stuck"]),
+			from: start,
+			to: end,
+			durableJobs: 0,
+		});
+		expect(queued.has("stuck")).toBe(true);
+		expect(
+			(checkedAt.get("stuck") ?? Number.POSITIVE_INFINITY) -
+				(stuck.startedAt + 60 * minute),
+		).toBeLessThanOrEqual(15 * minute);
+		const agedOut = rows.filter(
+			(row) =>
+				row.startedAt >= start - 7 * day && row.startedAt < end - 7 * day,
+		);
+		expect(agedOut.length).toBeGreaterThan(0);
+		for (const row of agedOut) expect(checkedAt.has(row.videoId)).toBe(true);
+		for (const row of rows) {
+			const qualified = row.startedAt + 60 * minute;
+			if (qualified < start || qualified > end - 15 * minute) continue;
+			expect(
+				(checkedAt.get(row.videoId) ?? Number.POSITIVE_INFINITY) - qualified,
+			).toBeLessThanOrEqual(15 * minute);
+		}
+	});
+
+	it("still reaches a finished recording in a burst while durable retries fill their budget", async () => {
+		const start = new Date("2026-10-01T00:00:00Z").getTime();
+		const burstAt = start - 2 * day;
+		const rows = emptyCaps(start - 8 * day, start + 6 * day);
+		for (let index = 0; index < 4; index++) {
+			rows.push({ videoId: `burst-empty-${index}`, startedAt: burstAt });
+		}
+		rows.push({ videoId: "burst-finished", startedAt: burstAt });
+		const { queued } = await simulate({
+			rows,
+			finished: new Set(["burst-finished"]),
+			from: start,
+			to: start + 6 * day,
+			durableJobs: 100,
+		});
+		expect(queued.has("burst-finished")).toBe(true);
+	});
+});

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

@@ -9,6 +9,7 @@ import { type User, Video } from "@cap/web-domain";
 import {
 	and,
 	asc,
+	desc,
 	eq,
 	gte,
 	inArray,
@@ -38,6 +39,7 @@ 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;
+export const DESKTOP_SEGMENTS_LEGACY_NEWEST_BATCH_SIZE = 5;
 
 const RECOVERABLE_UPLOAD_PHASES = [
 	"uploading",
@@ -253,35 +255,49 @@ export async function recoverStaleDesktopSegments({
 	const staleBefore = new Date(
 		now.getTime() - DESKTOP_SEGMENTS_RECOVERY_MIN_AGE_MS,
 	);
-	const legacy = await db()
-		.select({ videoId: videos.id, ownerId: videos.ownerId })
-		.from(videoUploads)
-		.innerJoin(videos, eq(videoUploads.videoId, videos.id))
-		.leftJoin(
-			videoProcessingJobs,
-			eq(videoUploads.videoId, videoProcessingJobs.videoId),
-		)
-		.where(
-			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),
-				or(
-					isNull(videoUploads.processingError),
-					notLike(
-						videoUploads.processingError,
-						`${DESKTOP_RECORDING_SOURCE_REUPLOAD_REQUIRED}:%`,
+	const listLegacy = (order: typeof asc, limit: number) =>
+		db()
+			.select({ videoId: videos.id, ownerId: videos.ownerId })
+			.from(videoUploads)
+			.innerJoin(videos, eq(videoUploads.videoId, videos.id))
+			.leftJoin(
+				videoProcessingJobs,
+				eq(videoUploads.videoId, videoProcessingJobs.videoId),
+			)
+			.where(
+				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),
+					or(
+						isNull(videoUploads.processingError),
+						notLike(
+							videoUploads.processingError,
+							`${DESKTOP_RECORDING_SOURCE_REUPLOAD_REQUIRED}:%`,
+						),
+					),
+					sql`JSON_UNQUOTE(JSON_EXTRACT(${videos.source}, '$.type')) = 'desktopSegments'`,
 				),
-				sql`JSON_UNQUOTE(JSON_EXTRACT(${videos.source}, '$.type')) = 'desktopSegments'`,
-			),
-		)
-		.orderBy(asc(videoUploads.updatedAt), asc(videoUploads.videoId))
-		.limit(remaining);
+			)
+			.orderBy(order(videoUploads.updatedAt), order(videoUploads.videoId))
+			.limit(limit);
+	// Legacy rows that never finish keep their oldest `updatedAt`, so the
+	// oldest-first pass checks each recording only shortly before it leaves
+	// the window. That pass keeps every slot it had; a few of the newest
+	// eligible rows are checked on top, so a recording is first checked soon
+	// after it qualifies.
+	const oldest = await listLegacy(asc, remaining);
+	const picked = new Set(oldest.map((candidate) => candidate.videoId));
+	const newest = (
+		await listLegacy(desc, DESKTOP_SEGMENTS_LEGACY_NEWEST_BATCH_SIZE)
+	).filter((candidate) => !picked.has(candidate.videoId));
+	const legacy = [...oldest, ...newest];
 	for (const candidate of legacy) {
 		try {
 			const result = await completeDesktopSegmentsManifestAndQueue({