diff --git a/apps/web/__tests__/unit/desktop-segments-recovery.test.ts b/apps/web/__tests__/unit/desktop-segments-recovery.test.ts index 6acb665694..ec0d246a98 100644 --- a/apps/web/__tests__/unit/desktop-segments-recovery.test.ts +++ b/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; + from: number; + to: number; + durableJobs: number; + }) { + const queued = new Set(); + 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(); + 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); + }); +}); diff --git a/apps/web/lib/desktop-segments-recovery.ts b/apps/web/lib/desktop-segments-recovery.ts index 3d93cdddc4..a1f9fe17da 100644 --- a/apps/web/lib/desktop-segments-recovery.ts +++ b/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({