feat(render-farm): export product recordings with GPU source transcodes

apps/render-farm/README.md

@@ -74,6 +74,10 @@ expires `hls/` and `jobs/` objects.
 | `RF_TOKEN` | required | Bearer token between clients, coordinator and workers |
 | `RF_S3_ENDPOINT`, `RF_S3_BUCKET`, `RF_S3_REGION` | required | Bucket holding recordings, outputs, HLS and the journal |
 | `RF_S3_IMDS` | off | Use the instance role instead of `RF_S3_ACCESS_KEY_ID`/`RF_S3_SECRET_ACCESS_KEY` |
+| `RF_MEDIA_S3_BUCKET` (`_REGION`, `_ENDPOINT`) | the `RF_S3_*` bucket | Bucket holding recordings and receiving exports; may belong to another account whose policy grants the farm's role. Objects are written `bucket-owner-full-control`; the journal stays in `RF_S3_BUCKET` |
+| `RF_CALLBACK_HOSTS` | none | Host suffixes a job's `callbackUrl` may use (e.g. `vercel.app,cap.so`) |
+| `RF_CALLBACK_SECRET` | `RF_TOKEN` | HMAC key for callback signatures |
+| `RF_TRANSCODE_ENCODER` | `h264_nvenc` | Encoder for source transcodes (`libx264` without a GPU) |
 | `RF_COORDINATOR_URL` | `http://127.0.0.1:8080` | Coordinator address (workers) and its advertised URL |
 | `RF_SLOTS` / `RF_AUDIO_SLOTS` | `5` / `4` | Render slots and audio lanes per worker (tuned on one L4 with 8 vCPUs) |
 | `RF_LOCAL_AUDIO_SLOTS` | `2` | Audio lanes on the coordinator |
@@ -92,6 +96,27 @@ expires `hls/` and `jobs/` objects.
 | `CAP_DECODER_READAHEAD` | `8` | Frames each decoder decodes ahead of the renderer |
 | `RF_HOT_SWAP` | off | Development: pull the engine, app and tuning from the bucket's `bin/` pointers |
 
+## Product integration
+
+A job can export a recording that lives in the product's bucket:
+
+- `sourceRoot` names the recording's folder (e.g. `<owner>/<video>/`); manifest
+  keys may point anywhere inside it, so a small render project (manifest,
+  project config, cursors) can sit beside the untouched source files.
+- `output: { key, hlsPrefix }` places the MP4 and HLS segments inside that
+  folder instead of `out/` and `hls/`.
+- `callbackUrl` receives the job's outcome once it is ready or failed, signed
+  `x-render-farm-signature: sha256=<hex HMAC of the body>`. `GET /jobs/:id`
+  reports `progress` (0-1), `hlsSegments` and a freshly signed `hlsUrl`.
+
+Browser recordings are WebM with no seek index and few keyframes, so chunks
+cannot start in the middle of them. A manifest entry with `transcodeFrom`
+names such a source; the coordinator first has a GPU slot re-encode it into
+the entry's `key` as H.264 with a keyframe every second (kept and reused for
+later exports). `POST /transcodes {sourceRoot, source, output}` starts one
+ahead of time (e.g. when the editor opens) and `GET /transcodes/:id` reports
+it.
+
 ## Development
 
 ```sh

apps/render-farm/src/coordinator.test.ts

@@ -1,5 +1,10 @@
 import { describe, expect, test } from "bun:test";
-import { randomUUID, timingSafeEqual } from "node:crypto";
+import {
+	createHash,
+	createHmac,
+	randomUUID,
+	timingSafeEqual,
+} from "node:crypto";
 import { readFileSync } from "node:fs";
 import type { Job, TaskState } from "./coordinator";
 import * as fmp4 from "./fmp4";
@@ -18,7 +23,12 @@ function harness() {
 	const watchdogs: (() => void)[] = [];
 	let putGate: Promise<void> | undefined;
 	let failures = 0;
+	const callbacks: { url: string; init: RequestInit }[] = [];
 	const s3 = {
+		async head(key: string) {
+			const value = objects.get(key);
+			return value ? { size: value.byteLength } : null;
+		},
 		async put(key: string, body: Uint8Array | string) {
 			writes.push(key);
 			await putGate;
@@ -53,6 +63,8 @@ function harness() {
 	const deps = {
 		timingSafeEqual,
 		randomUUID,
+		createHash,
+		createHmac,
 		...validate,
 		...fmp4,
 		...hls,
@@ -67,10 +79,16 @@ function harness() {
 			}
 		},
 		s3ConfigFromEnv: () => ({}),
+		mediaS3ConfigFromEnv: () => ({}),
 		ProbeEngine: class {},
 		Engine: class {},
 		process: {
-			env: { RF_TOKEN: "test", RF_LOCAL_AUDIO_SLOTS: "0", RF_HLS: "1" },
+			env: {
+				RF_TOKEN: "test",
+				RF_LOCAL_AUDIO_SLOTS: "0",
+				RF_HLS: "1",
+				RF_CALLBACK_HOSTS: "cap.test",
+			},
 		},
 		Bun: {
 			serve: (options: { fetch: typeof fetchHandler }) => {
@@ -85,6 +103,10 @@ function harness() {
 			return { unref() {} };
 		},
 		clearTimeout: () => {},
+		fetch: async (url: string, init: RequestInit) => {
+			callbacks.push({ url, init });
+			return new Response("ok");
+		},
 		console: { log() {}, warn() {}, error() {} },
 		mkdirSync: () => {},
 		rmSync: () => {},
@@ -122,6 +144,7 @@ function harness() {
 		...coordinator,
 		objects,
 		writes,
+		callbacks,
 		timers,
 		watchdogs,
 		fetch: (request: Request) => fetchHandler(request),
@@ -521,3 +544,173 @@ test("job acknowledgement waits for a planning receipt and receipt-only jobs res
 	expect(planned).toEqual([receipt.id, receipt.id]);
 	expect(h.jobs.get(receipt.id)?.status).toBe("planning");
 });
+
+function call(
+	h: ReturnType<typeof harness>,
+	path: string,
+	body?: Record<string, unknown>,
+) {
+	return h.fetch(
+		new Request(`http://test${path}`, {
+			method: body ? "POST" : "GET",
+			headers: {
+				authorization: "Bearer test",
+				"content-type": "application/json",
+			},
+			body: body ? JSON.stringify(body) : undefined,
+		}),
+	);
+}
+
+describe("transcodes", () => {
+	const request = {
+		sourceRoot: "owner/video/",
+		source: "owner/video/raw-upload.webm",
+		output: "owner/video/.recording/render/sources/display.mp4",
+	};
+
+	test("are queued once, go to a video slot first and settle on the worker's report", async () => {
+		const h = harness();
+		const created = (await (await call(h, "/transcodes", request)).json()) as {
+			id: string;
+			status: string;
+		};
+		expect(created.status).toBe("queued");
+		const again = (await (await call(h, "/transcodes", request)).json()) as {
+			id: string;
+		};
+		expect(again.id).toBe(created.id);
+		const work = (await (
+			await call(h, "/work", {
+				worker: "gpu-a",
+				slots: 1,
+				cpus: 8,
+				kinds: ["video"],
+			})
+		).json()) as { task: protocol.TranscodeTask };
+		expect(work.task).toMatchObject({
+			kind: "transcode",
+			source: request.source,
+			output: request.output,
+			attempt: 1,
+		});
+		const heartbeat = (await (
+			await call(h, "/heartbeat", {
+				worker: "gpu-b",
+				slots: 1,
+				cpus: 8,
+				running: [
+					{
+						taskId: work.task.taskId,
+						attempt: 1,
+						frames: 3,
+						total: 0,
+						elapsedMs: 10,
+					},
+				],
+			})
+		).json()) as { cancel: string[] };
+		expect(heartbeat.cancel).toContain(work.task.taskId);
+		await call(h, `/transcodes/${created.id}/done`, {
+			worker: "gpu-a",
+			attempt: 1,
+			size: 1234,
+		});
+		expect(
+			await (await call(h, `/transcodes/${created.id}`)).json(),
+		).toMatchObject({ status: "ready", size: 1234 });
+	});
+
+	test("reuse an output already in the bucket", async () => {
+		const h = harness();
+		h.objects.set(request.output, new Uint8Array(42));
+		expect(await (await call(h, "/transcodes", request)).json()).toMatchObject({
+			status: "ready",
+			size: 42,
+		});
+	});
+
+	test("must stay inside their source folder", async () => {
+		const h = harness();
+		for (const body of [
+			{ ...request, source: "other/raw-upload.webm" },
+			{ ...request, output: "owner/other/display.mp4" },
+			{ ...request, output: "owner/video/display.webm" },
+			{ ...request, sourceRoot: "owner/../" },
+		]) {
+			expect((await call(h, "/transcodes", body)).status).toBe(400);
+		}
+	});
+});
+
+describe("jobs for the product", () => {
+	test("write to the requested keys and call back signed when finished", async () => {
+		const h = harness();
+		h.setPlanner(async () => {});
+		const receipt = (await (
+			await call(h, "/jobs", {
+				recording: "owner/video/.recording/render/abc/project",
+				sourceRoot: "owner/video/",
+				output: {
+					key: "owner/video/.recording/render/abc/result.mp4",
+					hlsPrefix: "owner/video/.recording/render/abc/hls",
+				},
+				callbackUrl: "https://preview.cap.test/api/render-farm/callback",
+				reference: "video-1",
+			})
+		).json()) as { id: string };
+		const exported = h.jobs.get(receipt.id) as Job;
+		expect(exported.key).toBe("owner/video/.recording/render/abc/result.mp4");
+		expect(exported.hls?.prefix).toBe("owner/video/.recording/render/abc/hls");
+		exported.status = "ready";
+		exported.totalFrames = 60;
+		h.finish(exported);
+		const callback = h.callbacks[0];
+		expect(callback?.url).toBe(
+			"https://preview.cap.test/api/render-farm/callback",
+		);
+		const body = String(callback?.init.body);
+		expect(JSON.parse(body)).toMatchObject({
+			id: receipt.id,
+			reference: "video-1",
+			status: "ready",
+			key: "owner/video/.recording/render/abc/result.mp4",
+			durationSeconds: 2,
+		});
+		expect(
+			(callback?.init.headers as Record<string, string>)[
+				"x-render-farm-signature"
+			],
+		).toBe(`sha256=${createHmac("sha256", "test").update(body).digest("hex")}`);
+	});
+
+	test("refuse callbacks to hosts that are not allowed", async () => {
+		const h = harness();
+		h.setPlanner(async () => {});
+		const response = await call(h, "/jobs", {
+			recording: "recording",
+			callbackUrl: "https://attacker.example/hook",
+		});
+		expect(response.status).toBe(400);
+	});
+
+	test("report the share of frames rendered", async () => {
+		const h = harness();
+		const j = job();
+		h.jobs.set(j.id, j);
+		j.videoResults.set(0, {} as protocol.VideoResult);
+		const running = videoState(j);
+		running.task = { ...running.task, chunk: 1 } as protocol.VideoTask;
+		running.progress = {
+			frames: 15,
+			total: 30,
+			elapsedMs: 100,
+			at: 0,
+			advancedAt: 0,
+		};
+		const summary = (await (await call(h, `/jobs/${j.id}`)).json()) as {
+			progress: number;
+		};
+		expect(summary.progress).toBe(0.75);
+	});
+});