perf(media-server): GPU/NVENC farm support, lifecycle hardening and faster verify/upload

apps/media-server/Dockerfile

@@ -15,13 +15,15 @@ COPY apps/media-server/capcodec.md ./capcodec.md
 RUN bun test src/__tests__/lib/audio-quality-policy.test.ts src/__tests__/lib/audio-quality.integration.test.ts \
 	&& bun test src/__tests__/lib/audio-quality-formats.integration.test.ts src/__tests__/lib/audio-quality-benchmark.test.ts \
 	&& bun test src/__tests__/lib/audio-levels.integration.test.ts \
-	&& bun test src/__tests__/lib/drive-resumable-upload.test.ts src/__tests__/lib/storage-upload.test.ts src/__tests__/lib/container-memory.test.ts \
+	&& bun test src/__tests__/lib/drive-resumable-upload.test.ts src/__tests__/lib/storage-upload.test.ts src/__tests__/lib/container-memory.test.ts src/__tests__/lib/parallel-download.test.ts \
 	&& bun test src/__tests__/lib/media-size.test.ts src/__tests__/lib/media-probe.integration.test.ts \
 	&& bun test src/__tests__/lib/media-audio.integration.test.ts \
 	&& bun test src/__tests__/lib/recording-verification.integration.test.ts src/__tests__/lib/job-manager.test.ts \
-	&& bun test src/__tests__/lib/media-transfer.test.ts src/__tests__/lib/capcodec-encoder.test.ts \
+	&& bun test src/__tests__/lib/media-transfer.test.ts src/__tests__/lib/capcodec-encoder.test.ts src/__tests__/lib/nvenc-encoder.test.ts \
 	&& bun test src/__tests__/routes/recording-verification.test.ts \
 	&& bun test src/__tests__/routes/video.test.ts \
+	&& bun test src/__tests__/lib/lifecycle.test.ts src/__tests__/lib/temp-files.test.ts \
+	&& bun test src/__tests__/routes/lifecycle.test.ts \
 	&& bun test src/__tests__/lib/media-routes-real-world.integration.test.ts \
 	&& bun test src/__tests__/lib/media-video.integration.test.ts
 
@@ -37,6 +39,6 @@ ENV PORT=3456
 EXPOSE 3456
 
 HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
-	CMD ["bun", "-e", "fetch(`http://127.0.0.1:${process.env.PORT || 3456}/health`).then((response) => process.exit(response.ok ? 0 : 1)).catch(() => process.exit(1))"]
+	CMD ["bun", "-e", "fetch(`http://127.0.0.1:${process.env.PORT || 3456}/health`).then(async (response) => process.exit(response.ok || (await response.json().catch(() => ({}))).status === 'draining' ? 0 : 1)).catch(() => process.exit(1))"]
 
 CMD ["bun", "run", "src/index.ts"]

apps/media-server/src/__tests__/lib/job-manager.test.ts

@@ -6,7 +6,9 @@ import {
 	cleanupExpiredJobs,
 	createJob,
 	deleteJob,
+	extendJobLifetime,
 	getJob,
+	getJobTotals,
 	type JobProgress,
 	type RecordingWorkerAcknowledgement,
 	sendWebhook,
@@ -522,3 +524,83 @@ describe("job cleanup", () => {
 		expect(currentJob?.error).toContain("maximum lifetime of 60 minutes");
 	});
 });
+
+describe("job lifetime", () => {
+	test("extends the lifetime cap for long work without shortening it", () => {
+		const job = createTrackedJob("job-extended-lifetime");
+		const now = Date.now();
+		job.phase = "processing";
+		job.createdAt = now - 10 * 60 * 1000;
+
+		expect(extendJobLifetime(job.jobId, 60_000)).toBe(
+			job.createdAt + 60 * 60 * 1000,
+		);
+		const deadline = extendJobLifetime(job.jobId, 2 * 60 * 60 * 1000);
+		expect(deadline).toBeGreaterThanOrEqual(now + 2 * 60 * 60 * 1000);
+		expect(extendJobLifetime(job.jobId, 60_000)).toBe(deadline);
+
+		job.createdAt = now - 90 * 60 * 1000;
+		job.updatedAt = Date.now();
+		expect(cleanupExpiredJobs()).toBe(0);
+		expect(getJob(job.jobId)?.phase).toBe("processing");
+	});
+
+	test("keeps a hard upper bound on extended lifetimes", () => {
+		const job = createTrackedJob("job-capped-lifetime");
+		job.phase = "uploading";
+		job.createdAt = Date.now() - 10 * 60 * 1000;
+
+		expect(extendJobLifetime(job.jobId, 24 * 60 * 60 * 1000)).toBe(
+			job.createdAt + 4 * 60 * 60 * 1000,
+		);
+		expect(extendJobLifetime(job.jobId, Number.NaN)).toBeUndefined();
+		expect(extendJobLifetime(job.jobId, -1)).toBeUndefined();
+	});
+
+	test("expires a job past its extended deadline", () => {
+		const job = createTrackedJob("job-past-extended-lifetime");
+		const abortController = new AbortController();
+		updateJob(job.jobId, { phase: "processing", abortController });
+		job.createdAt = Date.now() - 10 * 60 * 1000;
+		extendJobLifetime(job.jobId, 2 * 60 * 60 * 1000);
+		job.createdAt -= 3 * 60 * 60 * 1000;
+		job.lifetimeDeadlineAt = Date.now() - 1;
+		job.updatedAt = Date.now();
+
+		expect(cleanupExpiredJobs()).toBe(1);
+		expect(job.phase).toBe("error");
+		expect(job.error).toContain("maximum lifetime");
+		expect(abortController.signal.aborted).toBe(true);
+	});
+
+	test("does not extend terminal jobs", () => {
+		const job = createTrackedJob("job-terminal-lifetime");
+		updateJob(job.jobId, { phase: "complete" });
+		expect(extendJobLifetime(job.jobId, 2 * 60 * 60 * 1000)).toBeUndefined();
+		expect(job.lifetimeDeadlineAt).toBeUndefined();
+	});
+});
+
+describe("job totals", () => {
+	test("counts created, terminal, and expired jobs", () => {
+		const before = getJobTotals();
+		const completed = createTrackedJob("job-total-complete");
+		const errored = createTrackedJob("job-total-error");
+		const cancelled = createTrackedJob("job-total-cancelled");
+		const stale = createTrackedJob("job-total-stale");
+		updateJob(completed.jobId, { phase: "complete" });
+		updateJob(completed.jobId, { phase: "error" });
+		updateJob(errored.jobId, { phase: "error" });
+		updateJob(cancelled.jobId, { phase: "cancelled" });
+		updateJob(stale.jobId, { phase: "processing" });
+		stale.updatedAt = Date.now() - 16 * 60 * 1000;
+		cleanupExpiredJobs();
+
+		const after = getJobTotals();
+		expect(after.created - before.created).toBe(4);
+		expect(after.completed - before.completed).toBe(1);
+		expect(after.errored - before.errored).toBe(2);
+		expect(after.cancelled - before.cancelled).toBe(1);
+		expect(after.expired - before.expired).toBe(1);
+	});
+});

apps/media-server/src/__tests__/lib/lifecycle.test.ts

@@ -0,0 +1,207 @@
+import { afterEach, describe, expect, test } from "bun:test";
+import {
+	beginRequest,
+	createShutdownHandler,
+	getDrainBlockers,
+	getIdleTimeoutSeconds,
+	getShutdownGraceMs,
+	isDraining,
+	isIdle,
+	restoreIdleTimeoutOnFirstChunk,
+	setDraining,
+	setRequestIdleTimeout,
+	trackBackgroundTask,
+	waitForDrain,
+} from "../../lib/lifecycle";
+
+const originalGrace = process.env.MEDIA_SERVER_SHUTDOWN_GRACE_MS;
+const originalIdle = process.env.MEDIA_SERVER_IDLE_TIMEOUT_SECONDS;
+
+function deferred() {
+	let resolve = () => {};
+	const promise = new Promise<void>((done) => {
+		resolve = done;
+	});
+	return { promise, resolve };
+}
+
+afterEach(() => {
+	setDraining(false);
+	if (originalGrace === undefined)
+		delete process.env.MEDIA_SERVER_SHUTDOWN_GRACE_MS;
+	else process.env.MEDIA_SERVER_SHUTDOWN_GRACE_MS = originalGrace;
+	if (originalIdle === undefined)
+		delete process.env.MEDIA_SERVER_IDLE_TIMEOUT_SECONDS;
+	else process.env.MEDIA_SERVER_IDLE_TIMEOUT_SECONDS = originalIdle;
+});
+
+describe("shutdown configuration", () => {
+	test("defaults to the immediate shutdown path", () => {
+		delete process.env.MEDIA_SERVER_SHUTDOWN_GRACE_MS;
+		expect(getShutdownGraceMs()).toBe(0);
+		process.env.MEDIA_SERVER_SHUTDOWN_GRACE_MS = "not-a-number";
+		expect(getShutdownGraceMs()).toBe(0);
+		process.env.MEDIA_SERVER_SHUTDOWN_GRACE_MS = "120000";
+		expect(getShutdownGraceMs()).toBe(120_000);
+	});
+
+	test("bounds the idle timeout to Bun's maximum", () => {
+		delete process.env.MEDIA_SERVER_IDLE_TIMEOUT_SECONDS;
+		expect(getIdleTimeoutSeconds()).toBe(255);
+		process.env.MEDIA_SERVER_IDLE_TIMEOUT_SECONDS = "90";
+		expect(getIdleTimeoutSeconds()).toBe(90);
+		process.env.MEDIA_SERVER_IDLE_TIMEOUT_SECONDS = "10000";
+		expect(getIdleTimeoutSeconds()).toBe(255);
+	});
+
+	test("ignores servers without per-request timeouts", () => {
+		const calls: Array<[Request, number]> = [];
+		const request = new Request("http://localhost/video/convert");
+		setRequestIdleTimeout(undefined, request, 0);
+		setRequestIdleTimeout(
+			{
+				timeout: (req: Request, seconds: number) => calls.push([req, seconds]),
+			},
+			request,
+			0,
+		);
+		expect(calls).toEqual([[request, 0]]);
+	});
+});
+
+describe("drain", () => {
+	test("waits for background work and in-flight requests", async () => {
+		const work = deferred();
+		const task = trackBackgroundTask(work.promise);
+		const endRequest = beginRequest();
+		expect(isIdle()).toBe(false);
+		expect(getDrainBlockers().backgroundTasks).toBeGreaterThanOrEqual(1);
+
+		const drained = waitForDrain(5_000);
+		await Bun.sleep(50);
+		work.resolve();
+		await task;
+		await Bun.sleep(50);
+		endRequest();
+
+		expect(await drained).toBe(true);
+		expect(isIdle()).toBe(true);
+	});
+
+	test("gives up when the grace period elapses", async () => {
+		const work = deferred();
+		const task = trackBackgroundTask(work.promise);
+		const startedAt = Date.now();
+
+		expect(await waitForDrain(300)).toBe(false);
+		expect(Date.now() - startedAt).toBeGreaterThanOrEqual(250);
+
+		work.resolve();
+		await task;
+	});
+
+	test("stops tracking rejected background work", async () => {
+		const failure = trackBackgroundTask(Promise.reject(new Error("failed")));
+		await failure.catch(() => {});
+		await Bun.sleep(0);
+		expect(await waitForDrain(1_000)).toBe(true);
+	});
+});
+
+describe("shutdown handler", () => {
+	test("keeps today's immediate shutdown when no grace period is set", async () => {
+		const work = deferred();
+		const task = trackBackgroundTask(work.promise);
+		let finalized = 0;
+		const shutdown = createShutdownHandler(
+			async () => {
+				finalized++;
+			},
+			() => 0,
+		);
+
+		await shutdown("SIGTERM");
+		await shutdown("SIGTERM");
+
+		expect(finalized).toBe(1);
+		expect(isDraining()).toBe(false);
+		work.resolve();
+		await task;
+	});
+
+	test("drains before finalizing when a grace period is set", async () => {
+		const work = deferred();
+		const task = trackBackgroundTask(work.promise);
+		let finalizedWhileBusy: boolean | undefined;
+		const shutdown = createShutdownHandler(
+			async () => {
+				finalizedWhileBusy = !isIdle();
+			},
+			() => 10_000,
+		);
+
+		const shuttingDown = shutdown("SIGTERM");
+		await Bun.sleep(50);
+		expect(isDraining()).toBe(true);
+		expect(finalizedWhileBusy).toBeUndefined();
+
+		work.resolve();
+		await task;
+		await shuttingDown;
+		expect(finalizedWhileBusy).toBe(false);
+	});
+
+	test("forces the immediate path on a second signal", async () => {
+		const work = deferred();
+		const task = trackBackgroundTask(work.promise);
+		let finalized = 0;
+		const shutdown = createShutdownHandler(
+			async () => {
+				finalized++;
+			},
+			() => 60_000,
+		);
+
+		const shuttingDown = shutdown("SIGTERM");
+		await Bun.sleep(50);
+		expect(finalized).toBe(0);
+		await shutdown("SIGINT");
+		await shuttingDown;
+		expect(finalized).toBe(1);
+
+		work.resolve();
+		await task;
+	});
+});
+
+describe("streamed response idle timeout", () => {
+	test("comes back once, on the first chunk, and cancel still reaches the source", async () => {
+		let restores = 0;
+		let cancelled = false;
+		let push: ReadableStreamDefaultController<Uint8Array> | undefined;
+		const source = new ReadableStream<Uint8Array>({
+			start(controller) {
+				push = controller;
+			},
+			cancel() {
+				cancelled = true;
+			},
+		});
+		const reader = restoreIdleTimeoutOnFirstChunk(source, () => {
+			restores++;
+		}).getReader();
+
+		const first = reader.read();
+		await Bun.sleep(10);
+		expect(restores).toBe(0);
+
+		push?.enqueue(new Uint8Array([1]));
+		expect((await first).value).toEqual(new Uint8Array([1]));
+		push?.enqueue(new Uint8Array([2]));
+		await reader.read();
+		expect(restores).toBe(1);
+
+		await reader.cancel();
+		expect(cancelled).toBe(true);
+	});
+});