Add resource-aware job throttling & health info

Expose resource and job state in the health route (activeVideoProcesses, canAcceptNewVideoProcess, resources). Adjust FFmpeg subprocess handling to catch spawn errors and decrement active process tracking on failure. Update tests to mock the new job-manager API (getMaxConcurrentVideoProcesses, getSystemResources) and remove mocks for the removed increment/decrement functions.

Author: richiemcilroy

apps/media-server/src/__tests__/routes/video.test.ts

@@ -311,11 +311,11 @@ describe("POST /video/process", () => {
 		mock.module("../../lib/job-manager", () => ({
 			canAcceptNewVideoProcess: () => false,
 			getActiveVideoProcessCount: jobManager.getActiveVideoProcessCount,
+			getMaxConcurrentVideoProcesses: jobManager.getMaxConcurrentVideoProcesses,
+			getSystemResources: jobManager.getSystemResources,
 			getAllJobs: jobManager.getAllJobs,
 			generateJobId: jobManager.generateJobId,
 			createJob: jobManager.createJob,
-			incrementActiveVideoProcesses: jobManager.incrementActiveVideoProcesses,
-			decrementActiveVideoProcesses: jobManager.decrementActiveVideoProcesses,
 			getJob: jobManager.getJob,
 			updateJob: jobManager.updateJob,
 			deleteJob: jobManager.deleteJob,
@@ -347,6 +347,8 @@ describe("POST /video/process", () => {
 		mock.module("../../lib/job-manager", () => ({
 			canAcceptNewVideoProcess: () => true,
 			getActiveVideoProcessCount: () => 0,
+			getMaxConcurrentVideoProcesses: () => 3,
+			getSystemResources: jobManager.getSystemResources,
 			getAllJobs: () => [],
 			generateJobId: () => "test-job-id",
 			createJob: () => ({
@@ -358,8 +360,6 @@ describe("POST /video/process", () => {
 				createdAt: new Date(),
 				updatedAt: new Date(),
 			}),
-			incrementActiveVideoProcesses: () => {},
-			decrementActiveVideoProcesses: () => {},
 			getJob: () => null,
 			updateJob: () => null,
 			deleteJob: () => {},
@@ -410,11 +410,11 @@ describe("GET /video/process/:jobId/status", () => {
 		mock.module("../../lib/job-manager", () => ({
 			canAcceptNewVideoProcess: () => true,
 			getActiveVideoProcessCount: () => 0,
+			getMaxConcurrentVideoProcesses: () => 3,
+			getSystemResources: jobManager.getSystemResources,
 			getAllJobs: () => [],
 			generateJobId: () => "test-job-id",
 			createJob: jobManager.createJob,
-			incrementActiveVideoProcesses: () => {},
-			decrementActiveVideoProcesses: () => {},
 			getJob: () => ({
 				jobId: "test-job-id",
 				videoId: "test-video",
@@ -459,11 +459,11 @@ describe("POST /video/process/:jobId/cancel", () => {
 		mock.module("../../lib/job-manager", () => ({
 			canAcceptNewVideoProcess: () => true,
 			getActiveVideoProcessCount: () => 0,
+			getMaxConcurrentVideoProcesses: () => 3,
+			getSystemResources: jobManager.getSystemResources,
 			getAllJobs: () => [],
 			generateJobId: () => "test-job-id",
 			createJob: jobManager.createJob,
-			incrementActiveVideoProcesses: () => {},
-			decrementActiveVideoProcesses: () => {},
 			getJob: () => null,
 			updateJob: () => null,
 			deleteJob: () => {},
@@ -488,11 +488,11 @@ describe("POST /video/process/:jobId/cancel", () => {
 		mock.module("../../lib/job-manager", () => ({
 			canAcceptNewVideoProcess: () => true,
 			getActiveVideoProcessCount: () => 0,
+			getMaxConcurrentVideoProcesses: () => 3,
+			getSystemResources: jobManager.getSystemResources,
 			getAllJobs: () => [],
 			generateJobId: () => "test-job-id",
 			createJob: jobManager.createJob,
-			incrementActiveVideoProcesses: () => {},
-			decrementActiveVideoProcesses: () => {},
 			getJob: () => ({
 				jobId: "test-job-id",
 				videoId: "test-video",
@@ -523,17 +523,16 @@ describe("POST /video/process/:jobId/cancel", () => {
 	});
 
 	test("successfully cancels running job", async () => {
-		const abortController = new AbortController();
 		let abortCalled = false;
 
 		mock.module("../../lib/job-manager", () => ({
 			canAcceptNewVideoProcess: () => true,
 			getActiveVideoProcessCount: () => 0,
+			getMaxConcurrentVideoProcesses: () => 3,
+			getSystemResources: jobManager.getSystemResources,
 			getAllJobs: () => [],
 			generateJobId: () => "test-job-id",
 			createJob: jobManager.createJob,
-			incrementActiveVideoProcesses: () => {},
-			decrementActiveVideoProcesses: () => {},
 			getJob: () => ({
 				jobId: "test-job-id",
 				videoId: "test-video",

apps/media-server/src/lib/ffmpeg.ts

@@ -255,27 +255,33 @@ export function extractAudioStream(
 	const opts = { ...DEFAULT_OPTIONS, ...options };
 	const timeout = options.timeoutMs ?? EXTRACT_TIMEOUT_MS;
 
-	const ffmpegArgs = [
-		"ffmpeg",
-		"-i",
-		videoUrl,
-		"-vn",
-		"-acodec",
-		opts.codec,
-		"-b:a",
-		opts.bitrate,
-		"-f",
-		"mp3",
-		"pipe:1",
-	];
-
-	const proc = registerSubprocess(
-		spawn({
-			cmd: ffmpegArgs,
-			stdout: "pipe",
-			stderr: "pipe",
-		}),
-	);
+	let proc: Subprocess;
+	try {
+		const ffmpegArgs = [
+			"ffmpeg",
+			"-i",
+			videoUrl,
+			"-vn",
+			"-acodec",
+			opts.codec,
+			"-b:a",
+			opts.bitrate,
+			"-f",
+			"mp3",
+			"pipe:1",
+		];
+
+		proc = registerSubprocess(
+			spawn({
+				cmd: ffmpegArgs,
+				stdout: "pipe",
+				stderr: "pipe",
+			}),
+		);
+	} catch (err) {
+		activeProcesses--;
+		throw err;
+	}
 
 	let timeoutId: ReturnType<typeof setTimeout> | undefined;
 	let cleaned = false;

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

@@ -1,3 +1,4 @@
+import os from "node:os";
 import type { Subprocess } from "bun";
 import type { TempFileHandle } from "./temp-files";
 
@@ -58,33 +59,94 @@ export interface Job {
 const jobs = new Map<string, Job>();
 const JOB_TTL_MS = 60 * 60 * 1000;
 
-let activeVideoProcesses = 0;
-const maxConcurrentVideoProcesses =
+const configuredMaxProcesses =
 	Number.parseInt(
-		process.env.MEDIA_SERVER_MAX_CONCURRENT_VIDEO_PROCESSES ?? "3",
+		process.env.MEDIA_SERVER_MAX_CONCURRENT_VIDEO_PROCESSES ?? "0",
 		10,
-	) || 3;
+	) || 0;
+
+const cpuCount = os.cpus().length;
+const totalMemoryMB = os.totalmem() / (1024 * 1024);
+
+const CPU_LOAD_THRESHOLD = 0.8;
+const MEMORY_FREE_THRESHOLD = 0.15;
+
+function isActivePhase(phase: JobPhase): boolean {
+	return phase !== "complete" && phase !== "error" && phase !== "cancelled";
+}
 
 export function getActiveVideoProcessCount(): number {
-	return activeVideoProcesses;
+	let count = 0;
+	for (const job of jobs.values()) {
+		if (isActivePhase(job.phase)) {
+			count++;
+		}
+	}
+	return count;
 }
 
 export function getMaxConcurrentVideoProcesses(): number {
-	return maxConcurrentVideoProcesses;
+	if (configuredMaxProcesses > 0) {
+		return configuredMaxProcesses;
+	}
+	return Math.max(1, Math.floor(cpuCount / 2));
 }
 
-export function canAcceptNewVideoProcess(): boolean {
-	return activeVideoProcesses < maxConcurrentVideoProcesses;
+export interface SystemResources {
+	cpuCount: number;
+	loadAvg1m: number;
+	cpuPressure: number;
+	totalMemoryMB: number;
+	freeMemoryMB: number;
+	memoryUsagePercent: number;
+	configuredMax: number;
+	effectiveMax: number;
+	throttleReason: string | null;
 }
 
-export function incrementActiveVideoProcesses(): void {
-	activeVideoProcesses++;
-}
+export function getSystemResources(): SystemResources {
+	const loadAvg1m = os.loadavg()[0];
+	const freeMemoryMB = os.freemem() / (1024 * 1024);
+	const cpuPressure = loadAvg1m / cpuCount;
+	const memoryUsagePercent = 1 - freeMemoryMB / totalMemoryMB;
+	const max = getMaxConcurrentVideoProcesses();
+
+	let effectiveMax = max;
+	let throttleReason: string | null = null;
 
-export function decrementActiveVideoProcesses(): void {
-	if (activeVideoProcesses > 0) {
-		activeVideoProcesses--;
+	if (cpuPressure > CPU_LOAD_THRESHOLD) {
+		effectiveMax = Math.max(
+			1,
+			Math.floor(max * (1 - (cpuPressure - CPU_LOAD_THRESHOLD))),
+		);
+		throttleReason = `CPU load ${cpuPressure.toFixed(2)} exceeds ${CPU_LOAD_THRESHOLD} threshold`;
 	}
+
+	if (memoryUsagePercent > 1 - MEMORY_FREE_THRESHOLD) {
+		const memMax = Math.max(1, Math.floor(max * (1 - memoryUsagePercent)));
+		if (memMax < effectiveMax) {
+			effectiveMax = memMax;
+			throttleReason = `Memory usage ${(memoryUsagePercent * 100).toFixed(0)}% exceeds ${((1 - MEMORY_FREE_THRESHOLD) * 100).toFixed(0)}% threshold`;
+		}
+	}
+
+	return {
+		cpuCount,
+		loadAvg1m,
+		cpuPressure,
+		totalMemoryMB: Math.round(totalMemoryMB),
+		freeMemoryMB: Math.round(freeMemoryMB),
+		memoryUsagePercent,
+		configuredMax: configuredMaxProcesses,
+		effectiveMax,
+		throttleReason,
+	};
+}
+
+export function canAcceptNewVideoProcess(): boolean {
+	const active = getActiveVideoProcessCount();
+	const resources = getSystemResources();
+	return active < resources.effectiveMax;
 }
 
 export function generateJobId(): string {
@@ -186,6 +248,12 @@ export function cleanupExpiredJobs(): number {
 
 	for (const [jobId, job] of jobs) {
 		if (now - job.updatedAt > JOB_TTL_MS) {
+			if (isActivePhase(job.phase)) {
+				console.warn(
+					`[job-manager] Cleaning up stuck job ${jobId} (phase=${job.phase}, age=${Math.round((now - job.createdAt) / 60000)}m)`,
+				);
+				job.abortController?.abort();
+			}
 			deleteJob(jobId);
 			cleaned++;
 		}

apps/media-server/src/routes/health.ts

@@ -1,5 +1,10 @@
 import { spawn } from "bun";
 import { Hono } from "hono";
+import {
+	canAcceptNewVideoProcess,
+	getActiveVideoProcessCount,
+	getSystemResources,
+} from "../lib/job-manager";
 
 const health = new Hono();
 
@@ -26,12 +31,17 @@ health.get("/", async (c) => {
 		}
 	} catch {}
 
+	const resources = getSystemResources();
+
 	return c.json({
 		status: ffmpegAvailable ? "ok" : "degraded",
 		ffmpeg: {
 			available: ffmpegAvailable,
 			version: ffmpegVersion,
 		},
+		activeVideoProcesses: getActiveVideoProcessCount(),
+		canAcceptNewVideoProcess: canAcceptNewVideoProcess(),
+		resources,
 	});
 });