fix: fence recording workers and surface stalled processing

Author: richiemcilroy

.github/workflows/recording-reliability.yml

@@ -0,0 +1,47 @@
+name: Recording Reliability
+
+on:
+  workflow_dispatch:
+  pull_request:
+    paths:
+      - "apps/web/**"
+      - "apps/media-server/**"
+      - "packages/database/**"
+      - "packages/web-*/**"
+      - "crates/recording/**"
+      - "pnpm-lock.yaml"
+      - ".github/workflows/recording-reliability.yml"
+  push:
+    branches: [main]
+    paths:
+      - "apps/web/**"
+      - "apps/media-server/**"
+      - "packages/database/**"
+      - "packages/web-*/**"
+      - "crates/recording/**"
+      - "pnpm-lock.yaml"
+      - ".github/workflows/recording-reliability.yml"
+
+permissions:
+  contents: read
+
+concurrency:
+  group: recording-reliability-${{ github.head_ref || github.ref_name }}
+  cancel-in-progress: true
+
+jobs:
+  recording-contract:
+    name: Recording ownership, preservation, and recovery
+    runs-on: ubuntu-24.04
+    timeout-minutes: 15
+    steps:
+      - uses: actions/checkout@v4
+      - uses: ./.github/actions/setup-js
+      - name: Verify recording contracts
+        working-directory: apps/web
+        run: |
+          pnpm exec vitest run \
+            __tests__/unit/desktop-recording-*.test.ts \
+            __tests__/unit/desktop-segments-*.test.ts \
+            __tests__/unit/finalize-desktop-recording.test.ts \
+            __tests__/unit/media-server-progress.test.ts

apps/web/__tests__/unit/desktop-recording-health-cron.test.ts

@@ -0,0 +1,101 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+
+const { recover, getHealth } = vi.hoisted(() => ({
+	recover: vi.fn(),
+	getHealth: vi.fn(),
+}));
+
+vi.mock("@/lib/desktop-segments-recovery", () => ({
+	recoverStaleDesktopSegments: recover,
+}));
+vi.mock("@/lib/desktop-recording-health", () => ({
+	getDesktopRecordingHealth: getHealth,
+}));
+
+import { GET } from "@/app/api/cron/finalize-stale-desktop-segments/route";
+
+const healthy = {
+	status: "healthy",
+	checkedAt: "2026-09-04T14:00:00.000Z",
+	scope: "unresolved",
+	stalledWorkers: 0,
+	stalledCommits: 0,
+	retryLoops: 0,
+	blockedCommittedSources: 0,
+	changedSources: 0,
+};
+
+function request(token = "test-cron-secret") {
+	return new Request(
+		"http://localhost/api/cron/finalize-stale-desktop-segments",
+		{
+			headers: { authorization: `Bearer ${token}` },
+		},
+	);
+}
+
+beforeEach(() => {
+	vi.stubEnv("CRON_SECRET", "test-cron-secret");
+	recover.mockResolvedValue({ checked: 1, statuses: {}, results: [] });
+	getHealth.mockResolvedValue(healthy);
+	vi.spyOn(console, "error").mockImplementation(() => {});
+});
+
+afterEach(() => vi.unstubAllEnvs());
+
+describe("recording recovery health reporting", () => {
+	it("does not inspect or recover recordings without cron authentication", async () => {
+		expect((await GET(request("wrong"))).status).toBe(401);
+		expect(recover).not.toHaveBeenCalled();
+		expect(getHealth).not.toHaveBeenCalled();
+	});
+
+	it("does not classify incomplete uploads as processing incidents", async () => {
+		recover.mockResolvedValue({
+			checked: 2,
+			statuses: { "source-incomplete": 2 },
+			results: [],
+		});
+		const response = await GET(request());
+		expect(response.status).toBe(200);
+		expect((await response.json()).health).toEqual(healthy);
+		expect(console.error).not.toHaveBeenCalled();
+	});
+
+	it("runs recovery before surfacing persistent retry loops as a failed cron", async () => {
+		getHealth.mockResolvedValue({
+			...healthy,
+			status: "degraded",
+			retryLoops: 2,
+		});
+		const response = await GET(request());
+		expect(recover).toHaveBeenCalledOnce();
+		expect(recover.mock.invocationCallOrder[0]).toBeLessThan(
+			getHealth.mock.invocationCallOrder[0] ?? 0,
+		);
+		expect(response.status).toBe(503);
+		expect(await response.json()).toMatchObject({
+			success: false,
+			health: { retryLoops: 2 },
+		});
+		expect(console.error).toHaveBeenCalledWith(
+			"[recording-health] Processing needs attention",
+			expect.objectContaining({ retryLoops: 2, recoveryFailures: 0 }),
+		);
+	});
+
+	it("surfaces failed recovery even when no aged jobs remain", async () => {
+		recover.mockResolvedValue({
+			checked: 1,
+			statuses: { failed: 1 },
+			results: [],
+		});
+		expect((await GET(request())).status).toBe(503);
+	});
+
+	it("does not report healthy when the health query fails", async () => {
+		getHealth.mockRejectedValue(new Error("Database unavailable"));
+		await expect(GET(request())).rejects.toThrow("Database unavailable");
+		expect(recover).toHaveBeenCalledOnce();
+	});
+});

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

@@ -0,0 +1,48 @@
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+const { select, from, where } = vi.hoisted(() => ({
+	select: vi.fn(),
+	from: vi.fn(),
+	where: vi.fn(),
+}));
+
+vi.mock("@cap/database", () => ({ db: () => ({ select }) }));
+
+import { getDesktopRecordingHealth } from "@/lib/desktop-recording-health";
+
+const counts = {
+	stalledWorkers: 0,
+	stalledCommits: 0,
+	retryLoops: 0,
+	blockedCommittedSources: 0,
+	changedSources: 0,
+};
+
+beforeEach(() => {
+	select.mockReturnValue({ from });
+	from.mockReturnValue({ where });
+	where.mockResolvedValue([counts]);
+});
+
+describe("recording health", () => {
+	it("returns aggregate counts without recording identities or content", async () => {
+		where.mockResolvedValue([{ ...counts, retryLoops: 2, ownerId: "private" }]);
+		expect(await getDesktopRecordingHealth()).toMatchObject({
+			status: "degraded",
+			retryLoops: 2,
+		});
+		expect(await getDesktopRecordingHealth()).not.toHaveProperty("ownerId");
+	});
+
+	it.each([
+		undefined,
+		{ ...counts, retryLoops: Number.NaN },
+		{ ...counts, retryLoops: -1 },
+	])(
+		"fails closed when the database does not return valid health counts",
+		async (row) => {
+			where.mockResolvedValue(row ? [row] : []);
+			await expect(getDesktopRecordingHealth()).rejects.toThrow();
+		},
+	);
+});

apps/web/__tests__/unit/desktop-recording-job-status.test.ts

@@ -16,6 +16,115 @@ const input = {
 afterEach(() => vi.unstubAllGlobals());
 
 describe("recording job completion reconciliation", () => {
+	it("does not treat a replica-local 404 as worker death", async () => {
+		const fetcher = vi
+			.fn()
+			.mockResolvedValue(new Response(null, { status: 404 }));
+		vi.stubGlobal("fetch", fetcher);
+		expect(await observeDesktopRecordingJob(input)).toEqual({
+			status: "unavailable",
+			delivered: false,
+		});
+		expect(fetcher).toHaveBeenCalledOnce();
+	});
+
+	it("identifies workers whose lease can only be renewed by their own callback", async () => {
+		vi.stubGlobal(
+			"fetch",
+			vi.fn().mockResolvedValue(
+				Response.json({
+					jobId: "job",
+					videoId: "video",
+					phase: "processing",
+					recordingWorker: { version: 1, action: "progress", sequence: 3 },
+				}),
+			),
+		);
+		expect(await observeDesktopRecordingJob(input)).toEqual({
+			status: "active",
+			delivered: false,
+			workerProtocol: 1,
+		});
+	});
+
+	it.each([
+		{ success: true },
+		{
+			recordingWorker: {
+				version: 1,
+				status: "accepted",
+				generation: "generation",
+				attemptId: "attempt",
+				jobId: "other",
+				sequence: 4,
+			},
+		},
+		{
+			recordingWorker: {
+				version: 1,
+				status: "stale",
+				generation: "generation",
+				attemptId: "attempt",
+				jobId: "job",
+				sequence: 4,
+			},
+		},
+	])(
+		"requires an exact terminal acknowledgement for owned workers",
+		async (ack) => {
+			const fetcher = vi
+				.fn()
+				.mockResolvedValueOnce(
+					Response.json({
+						jobId: "job",
+						videoId: "video",
+						generation: "generation",
+						attemptId: "attempt",
+						phase: "complete",
+						recordingWorker: { version: 1, action: "progress", sequence: 4 },
+					}),
+				)
+				.mockResolvedValueOnce(Response.json(ack));
+			vi.stubGlobal("fetch", fetcher);
+			expect(await observeDesktopRecordingJob(input)).toEqual({
+				status: "terminal",
+				delivered: false,
+			});
+		},
+	);
+
+	it("accepts an exact redelivery acknowledgement after the completion ACK was lost", async () => {
+		const fence = {
+			generation: "generation",
+			attemptId: "attempt",
+			jobId: "job",
+		};
+		vi.stubGlobal(
+			"fetch",
+			vi
+				.fn()
+				.mockResolvedValueOnce(
+					Response.json({
+						...fence,
+						videoId: "video",
+						phase: "complete",
+						recordingWorker: { version: 1, action: "progress", sequence: 4 },
+					}),
+				)
+				.mockResolvedValueOnce(
+					Response.json({
+						recordingWorker: {
+							...fence,
+							version: 1,
+							status: "accepted",
+							sequence: 4,
+						},
+					}),
+				),
+		);
+		expect(await reconcileDesktopRecordingJob(input)).toBe(true);
+	});
+
 	it("redelivers a lost completion through the validated webhook", async () => {
 		const proof = {
 			request: { version: 1 },

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

@@ -7,6 +7,7 @@ import {
 	type DesktopRecordingJob,
 	ensureSegmentProcessingJob,
 	getDesktopRecordingRetryDelay,
+	getDesktopRecordingWorkerCheckpoint,
 	heartbeatAttempt,
 	initializeSourceCommitCheckpoint,
 	isDesktopRecordingJobRecoverable,
@@ -270,6 +271,60 @@ describe("durable recording job ownership", () => {
 		expect(rows.jobs?.[0]?.attemptCount).toBe(1);
 	});
 
+	it("keeps physical worker leases private and clears their checkpoint for a new attempt", async () => {
+		const first = await createAttempt();
+		expect(await persistCommittedSource(first, source)).toBe(true);
+		const row = rows.jobs?.[0];
+		if (!row) throw new Error("Missing processing job");
+		Object.assign(row, {
+			remoteJobId: "physical-worker",
+			output: {
+				version: 1,
+				kind: "recording-worker",
+				generation: first.generation,
+				attemptId: first.attemptId,
+				jobId: "physical-worker",
+				sequence: 5,
+				phase: "processing",
+				progress: 60,
+				payloadSha256: "d".repeat(64),
+				updatedAt: now.toISOString(),
+				stateChangedAt: now.toISOString(),
+			},
+		});
+		expect(await heartbeatAttempt(first)).toBe(false);
+		vi.setSystemTime(new Date(now.getTime() + 6 * 60_000));
+		const next = await claimProcessingAttempt({
+			videoId,
+			generation: first.generation,
+		});
+		expect(next).toMatchObject({ source, remoteJobId: null, output: null });
+		expect(next?.attemptId).not.toBe(first.attemptId);
+	});
+
+	it("rejects a worker checkpoint transplanted from another physical attempt", async () => {
+		const first = await createAttempt();
+		expect(() =>
+			getDesktopRecordingWorkerCheckpoint({
+				...first,
+				remoteJobId: "current-worker",
+				output: {
+					version: 1,
+					kind: "recording-worker",
+					generation: first.generation,
+					attemptId: first.attemptId,
+					jobId: "other-worker",
+					sequence: 0,
+					phase: "queued",
+					progress: 0,
+					payloadSha256: "d".repeat(64),
+					updatedAt: now.toISOString(),
+					stateChangedAt: now.toISOString(),
+				},
+			}),
+		).toThrow("does not match its owner");
+	});
+
 	it("rejects stale attempt heartbeats, remote jobs, and errors after a retry takes over", async () => {
 		const old = await createAttempt();
 		vi.setSystemTime(new Date(now.getTime() + 6 * 60_000));