fix: batch recording source checkpoints without losing retry safety

Author: richiemcilroy

apps/web/__tests__/fixtures/source-commit-timings.json

@@ -0,0 +1,67 @@
+[
+	{
+		"name": "long-s3",
+		"terminalResult": "ready",
+		"baselineWallMs": 919500,
+		"expectedBatchedSteps": 31,
+		"pageDurationsMs": [
+			464, 462, 512, 404, 577, 373, 517, 384, 672, 368, 458, 340, 592, 372, 516,
+			422, 841, 477, 559, 358, 600, 423, 508, 369, 621, 353, 449, 384, 546, 694,
+			442, 508, 837, 338, 439, 334, 596, 410, 436, 346, 682, 377, 442, 388, 557,
+			375, 484, 381, 757, 433, 483, 333, 564, 368, 510, 362, 662, 340, 418, 351,
+			538, 490, 512, 394, 928, 364, 429, 379, 547, 362, 449, 382, 668, 373, 436,
+			387, 550, 363, 442, 377, 731, 372, 488, 398, 578, 429, 529, 344, 625, 532,
+			469, 387, 576, 365, 501, 415, 901, 375, 523, 411, 668, 894, 496, 438, 695,
+			414, 480, 359, 687, 393, 493, 401, 781, 439, 531, 422, 590, 416, 461, 378,
+			702, 369, 448, 370, 549, 356, 477, 408, 1058, 383, 513, 417, 524, 368,
+			470, 378, 653, 363, 589, 369, 545, 360, 495, 392, 787, 387, 483, 358, 621,
+			448, 482, 377, 638, 345, 701, 417, 634, 381, 471, 504, 850, 415, 717, 489,
+			682, 1794, 1821, 1800, 1905, 1693, 1710, 1671, 2152, 1734, 1919, 1649,
+			1948, 1642, 1776, 1731, 2183, 1672, 1965, 1607, 1773, 1521, 1634, 1626,
+			1744, 1566, 1607, 1470, 1664, 1640, 1744, 1500, 2105, 1534, 1659, 1452,
+			1837, 1451, 1637, 1592, 1787, 1608, 1671, 1607, 1673, 1737, 1614, 1550,
+			1930, 1772, 1613, 1602, 1758, 1737, 1652, 1674, 1912, 1743, 1722, 1880,
+			1895, 1713, 1642, 1600, 2281, 1694, 1926, 1797, 2012, 1641, 1813, 1555,
+			1969, 1804, 1870, 1570, 2104, 1655, 1985, 1704, 1935, 1824, 2122, 1694,
+			1829, 1577, 1654, 1588, 1850, 1661, 1899, 1723, 1767, 1604, 1615, 1625,
+			2481, 1714, 1571, 1655, 1710, 1596, 1627, 1557, 1889, 1338, 1522, 1584,
+			1698, 1703, 1662, 1835, 1748, 1770, 1595, 1393, 1699, 1676, 1533, 1423,
+			1954, 1445, 1595, 1551, 1611, 1457, 1709, 1453, 2335, 1532, 1519, 1394,
+			1512, 1335, 1492, 1405, 1826, 1765, 1479, 1385, 1626, 1342, 1513, 1438,
+			1688, 1434, 1761, 1545, 1533, 1890, 1659, 1571, 1927, 1325, 1745, 1550,
+			1633, 1423, 1592, 1556, 1703, 1459, 1362, 1337, 1517, 1178, 786, 719, 611,
+			702, 739, 693, 630, 966, 800, 713, 676, 920, 872, 695, 678, 684, 680, 644,
+			924, 775, 684, 791, 634, 815, 724, 775, 688, 690, 631, 745, 973, 643, 636,
+			691, 686, 751, 671, 762, 680, 838, 629, 608, 640, 929, 805, 708, 648, 704,
+			677, 643, 729, 705, 675, 762, 940, 869, 689, 1050, 799, 701, 696, 632,
+			768, 711, 723, 723, 715, 695, 660, 642, 671, 642, 735, 1002, 705, 694,
+			668, 700, 830, 686, 640, 709, 678, 656, 716, 649, 681, 733, 750, 644, 828,
+			701, 733, 759, 651, 873, 657, 664, 675, 667, 938, 646, 764, 756, 609,
+			1117, 820, 627, 619, 660, 585, 622, 744, 763, 741, 769, 641, 765, 794,
+			736, 779, 687, 942, 665, 635, 752, 618, 686, 809, 780, 623, 914, 572, 620,
+			627, 708, 579, 729, 713, 604, 613, 592, 629, 555, 672, 644, 648, 646, 588,
+			587, 799, 681, 644, 666, 564, 590, 795, 772, 686, 444, 485, 464, 493, 542
+		]
+	},
+	{
+		"name": "short-drive-first",
+		"terminalResult": "ready",
+		"baselineWallMs": 29925,
+		"expectedBatchedSteps": 2,
+		"pageDurationsMs": [4693, 2796, 10156, 1521, 8332]
+	},
+	{
+		"name": "short-drive-retry",
+		"terminalResult": "ready",
+		"baselineWallMs": 26610,
+		"expectedBatchedSteps": 2,
+		"pageDurationsMs": [4867, 2914, 8932, 2810, 4612]
+	},
+	{
+		"name": "incomplete-source",
+		"terminalResult": "source-blocked",
+		"baselineWallMs": 185,
+		"expectedBatchedSteps": 1,
+		"pageDurationsMs": [185]
+	}
+]

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

@@ -5,7 +5,10 @@ import type {
 	DesktopRecordingAttempt,
 	DesktopRecordingJob,
 } from "@/lib/desktop-recording-jobs";
+import type { DesktopRecordingSourceCheckpoint } from "@/lib/desktop-recording-source-checkpoint";
+import type { RecordingVerification } from "@/lib/desktop-recording-verification";
 import { MediaProcessingBudgetError } from "@/lib/media-processing-budget";
+import sourceCommitTimings from "../fixtures/source-commit-timings.json";
 
 const mocks = vi.hoisted(() => ({
 	state: vi.fn(),
@@ -465,8 +468,262 @@ describe("short durable processing polls", () => {
 	});
 });
 
+describe("bounded source commitment steps", () => {
+	beforeEach(() => {
+		withCurrent({ source: null, state: "committing" });
+	});
+
+	function advanceCheckpoint(
+		_video: unknown,
+		checkpoint: DesktopRecordingSourceCheckpoint,
+	) {
+		return { checkpoint: { ...checkpoint, revision: checkpoint.revision + 1 } };
+	}
+
+	it.each(sourceCommitTimings)(
+		"replays the $name production timings without repeating or skipping a page",
+		async ({ pageDurationsMs, expectedBatchedSteps, terminalResult }) => {
+			const revisions: number[] = [];
+			mocks.commitSource.mockImplementation(async (_video, checkpoint) => {
+				const duration = pageDurationsMs[checkpoint.revision];
+				if (duration === undefined) throw new Error("Unexpected source page");
+				vi.setSystemTime(Date.now() + duration);
+				revisions.push(checkpoint.revision);
+				if (checkpoint.revision + 1 < pageDurationsMs.length)
+					return advanceCheckpoint(_video, checkpoint);
+				if (terminalResult === "source-blocked")
+					throw Object.assign(new Error("Manifest is incomplete"), {
+						code: "source-incomplete",
+					});
+				return { source };
+			});
+			let batches = 0;
+			let result: Awaited<ReturnType<typeof commitDesktopRecordingAttempt>>;
+			do {
+				if (++batches > pageDurationsMs.length)
+					throw new Error("Source preparation did not terminate");
+				result = await commitDesktopRecordingAttempt(fixture);
+			} while (result === "progress");
+			expect(result).toBe(terminalResult);
+			expect(batches).toBe(expectedBatchedSteps);
+			expect(revisions).toEqual(pageDurationsMs.map((_, index) => index));
+			expect(mocks.saveCheckpoint).toHaveBeenCalledTimes(
+				pageDurationsMs.length - 1,
+			);
+			expect(mocks.persist).toHaveBeenCalledTimes(
+				terminalResult === "ready" ? 1 : 0,
+			);
+		},
+	);
+
+	it("persists every page before advancing and commits only once after a lost step response", async () => {
+		const revisions: number[] = [];
+		mocks.commitSource.mockImplementation(async (_video, checkpoint) => {
+			expect(current?.output).toEqual(
+				checkpoint.revision === 0 ? null : checkpoint,
+			);
+			revisions.push(checkpoint.revision);
+			return checkpoint.revision === 4
+				? { source }
+				: advanceCheckpoint(_video, checkpoint);
+		});
+		expect(await commitDesktopRecordingAttempt(fixture)).toBe("ready");
+		expect(revisions).toEqual([0, 1, 2, 3, 4]);
+		expect(mocks.saveCheckpoint).toHaveBeenCalledTimes(4);
+		expect(await commitDesktopRecordingAttempt(fixture)).toBe("ready");
+		expect(mocks.commitSource).toHaveBeenCalledTimes(5);
+		expect(mocks.persist).toHaveBeenCalledOnce();
+		expect(mocks.fetch).not.toHaveBeenCalled();
+	});
+
+	it.each([
+		[0, 32],
+		[-1, 32],
+		[1_000, 15],
+		[14_999, 2],
+		[15_000, 1],
+		[180_000, 1],
+	])("yields after bounded work with %i ms pages", async (duration, pages) => {
+		mocks.commitSource.mockImplementation(async (_video, checkpoint) => {
+			vi.setSystemTime(Date.now() + duration);
+			return advanceCheckpoint(_video, checkpoint);
+		});
+		expect(await commitDesktopRecordingAttempt(fixture)).toBe("progress");
+		expect(mocks.commitSource).toHaveBeenCalledTimes(pages);
+		expect(mocks.saveCheckpoint).toHaveBeenCalledTimes(pages);
+		expect(current?.output).toMatchObject({ revision: pages });
+		expect(mocks.persist).not.toHaveBeenCalled();
+	});
+
+	it("includes checkpoint persistence latency in the batch budget", async () => {
+		mocks.commitSource.mockImplementation(advanceCheckpoint);
+		mocks.saveCheckpoint.mockImplementation(async (_attempt, checkpoint) => {
+			withCurrent({ output: checkpoint });
+			vi.setSystemTime(Date.now() + 15_000);
+			return true;
+		});
+		expect(await commitDesktopRecordingAttempt(fixture)).toBe("progress");
+		expect(mocks.commitSource).toHaveBeenCalledOnce();
+		expect(current?.output).toMatchObject({ revision: 1 });
+	});
+
+	it("resumes from the last saved page when a later storage operation fails", async () => {
+		mocks.commitSource
+			.mockImplementationOnce(advanceCheckpoint)
+			.mockImplementationOnce(advanceCheckpoint)
+			.mockRejectedValueOnce(new Error("Storage unavailable"));
+		expect(await commitDesktopRecordingAttempt(fixture)).toBe("progress");
+		expect(current?.output).toMatchObject({ revision: 2 });
+		expect(mocks.blocked).not.toHaveBeenCalled();
+		expect(await commitDesktopRecordingAttempt(fixture)).toBe("ready");
+		expect(
+			mocks.commitSource.mock.calls.map((call) => call[1].revision),
+		).toEqual([0, 1, 2, 2]);
+		expect(mocks.persist).toHaveBeenCalledOnce();
+	});
+
+	it("preserves a fresh durable retry budget after each saved page", async () => {
+		const failedRevisions = new Set<number>();
+		mocks.commitSource.mockImplementation(async (_video, checkpoint) => {
+			if (
+				checkpoint.revision > 0 &&
+				!failedRevisions.has(checkpoint.revision)
+			) {
+				failedRevisions.add(checkpoint.revision);
+				throw new Error("Intermittent storage failure");
+			}
+			return checkpoint.revision === 5
+				? { source }
+				: advanceCheckpoint(_video, checkpoint);
+		});
+		for (let revision = 1; revision <= 5; revision++) {
+			expect(await commitDesktopRecordingAttempt(fixture)).toBe("progress");
+			expect(current?.output).toMatchObject({ revision });
+		}
+		expect(await commitDesktopRecordingAttempt(fixture)).toBe("ready");
+		expect(mocks.saveCheckpoint).toHaveBeenCalledTimes(5);
+		expect(mocks.retry).not.toHaveBeenCalled();
+	});
+
+	it("throws persistent failures when the new batch cannot make progress", async () => {
+		mocks.commitSource.mockImplementationOnce(advanceCheckpoint);
+		mocks.commitSource.mockRejectedValue(new Error("Storage unavailable"));
+		expect(await commitDesktopRecordingAttempt(fixture)).toBe("progress");
+		await expect(commitDesktopRecordingAttempt(fixture)).rejects.toThrow(
+			"Storage unavailable",
+		);
+		expect(current?.output).toMatchObject({ revision: 1 });
+		expect(mocks.commitSource).toHaveBeenCalledTimes(3);
+	});
+
+	it.each([false, true])(
+		"reconciles a failed checkpoint response with database commit=%s",
+		async (committed) => {
+			mocks.commitSource.mockImplementationOnce(advanceCheckpoint);
+			mocks.saveCheckpoint.mockImplementationOnce(
+				async (_attempt, checkpoint) => {
+					if (committed) withCurrent({ output: checkpoint });
+					throw new Error("Database response lost");
+				},
+			);
+			await expect(commitDesktopRecordingAttempt(fixture)).rejects.toThrow(
+				"Database response lost",
+			);
+			expect(mocks.commitSource).toHaveBeenCalledOnce();
+			expect(await commitDesktopRecordingAttempt(fixture)).toBe("ready");
+			expect(mocks.commitSource.mock.calls[1]?.[1]).toMatchObject({
+				revision: committed ? 1 : 0,
+			});
+		},
+	);
+
+	it("stops immediately when checkpoint persistence loses its ownership fence", async () => {
+		mocks.commitSource.mockImplementation(advanceCheckpoint);
+		mocks.saveCheckpoint.mockResolvedValueOnce(false);
+		expect(await commitDesktopRecordingAttempt(fixture)).toBe("superseded");
+		expect(mocks.commitSource).toHaveBeenCalledOnce();
+		expect(mocks.persist).not.toHaveBeenCalled();
+	});
+
+	it.each(["deleted", "replaced", "lease-expired"])(
+		"stops between pages when the job is %s",
+		async (change) => {
+			mocks.commitSource.mockImplementationOnce(advanceCheckpoint);
+			mocks.saveCheckpoint.mockImplementationOnce(
+				async (_attempt, checkpoint) => {
+					withCurrent({ output: checkpoint });
+					if (change === "deleted") current = null;
+					else if (change === "replaced")
+						withCurrent({ attemptId: "replacement" });
+					else mocks.checkpoint.mockResolvedValueOnce(null);
+					return true;
+				},
+			);
+			expect(await commitDesktopRecordingAttempt(fixture)).toBe("superseded");
+			expect(mocks.commitSource).toHaveBeenCalledOnce();
+			expect(mocks.persist).not.toHaveBeenCalled();
+		},
+	);
+
+	it("refreshes the video and late verification before the next page", async () => {
+		const verification: RecordingVerification = {
+			version: 1,
+			artifact: { kind: "segments", manifestSha256: source.manifestSha256 },
+			requiredAudio: true,
+		};
+		const updatedVideo = { id: "video", storageIntegrationId: "updated" };
+		mocks.commitSource.mockImplementationOnce(async (_video, checkpoint) => {
+			withCurrent({ verification });
+			mocks.databaseRows.mockResolvedValueOnce([updatedVideo]);
+			return advanceCheckpoint(_video, checkpoint);
+		});
+		expect(await commitDesktopRecordingAttempt(fixture)).toBe("ready");
+		expect(mocks.commitSource.mock.calls[0]?.[2]).toBeUndefined();
+		expect(mocks.commitSource.mock.calls[1]?.slice(0, 3)).toEqual([
+			updatedVideo,
+			expect.objectContaining({ revision: 1 }),
+			verification,
+		]);
+	});
+
+	it("keeps heartbeat ownership checks on later pages", async () => {
+		mocks.commitSource
+			.mockImplementationOnce(advanceCheckpoint)
+			.mockImplementationOnce(
+				async (_video, _checkpoint, _verification, pulse) => {
+					await pulse();
+					return { source };
+				},
+			);
+		mocks.heartbeat.mockResolvedValueOnce(false);
+		expect(await commitDesktopRecordingAttempt(fixture)).toBe("progress");
+		expect(current?.output).toMatchObject({ revision: 1 });
+		expect(mocks.persist).not.toHaveBeenCalled();
+		mocks.checkpoint.mockResolvedValueOnce(null);
+		expect(await commitDesktopRecordingAttempt(fixture)).toBe("superseded");
+		expect(mocks.commitSource).toHaveBeenCalledTimes(2);
+	});
+
+	it.each([
+		"source-incomplete",
+		"source-missing",
+		"source-changed",
+		"source-invalid",
+	])("preserves %s rejection after earlier pages were saved", async (code) => {
+		mocks.commitSource
+			.mockImplementationOnce(advanceCheckpoint)
+			.mockRejectedValueOnce(Object.assign(new Error(code), { code }));
+		expect(await commitDesktopRecordingAttempt(fixture)).toBe("source-blocked");
+		expect(mocks.blocked).toHaveBeenCalledWith(
+			expect.objectContaining({ errorCode: code }),
+		);
+		expect(mocks.persist).not.toHaveBeenCalled();
+		expect(mocks.fetch).not.toHaveBeenCalled();
+	});
+});
+
 describe("source commitment and media request compatibility", () => {
-	it("checkpoints each source batch in a separate durable step before processing", async () => {
+	it("checkpoints each source page before processing", async () => {
 		withCurrent({ source: null, state: "committing", leaseExpiresAt: null });
 		mocks.commitSource.mockImplementationOnce(async (_video, checkpoint) => ({
 			checkpoint: { ...checkpoint, revision: 1, phase: "copy" },