Improve job cleanup and add force-cleanup API

Author: richiemcilroy

apps/media-server/src/app.ts

@@ -28,6 +28,8 @@ app.get("/", (c) => {
 			"/video/process",
 			"/video/process/:jobId/status",
 			"/video/process/:jobId/cancel",
+			"/video/cleanup",
+			"/video/force-cleanup",
 		],
 	});
 });

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

@@ -189,8 +189,13 @@ export async function downloadVideoToTemp(
 	);
 
 	try {
+		const timeoutSignal = AbortSignal.timeout(DOWNLOAD_TIMEOUT_MS);
+		const combinedSignal = abortSignal
+			? AbortSignal.any([abortSignal, timeoutSignal])
+			: timeoutSignal;
+
 		const response = await fetch(videoUrl, {
-			signal: abortSignal ?? AbortSignal.timeout(DOWNLOAD_TIMEOUT_MS),
+			signal: combinedSignal,
 		});
 
 		console.log(

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

@@ -58,6 +58,8 @@ export interface Job {
 
 const jobs = new Map<string, Job>();
 const JOB_TTL_MS = 60 * 60 * 1000;
+const STALE_JOB_MS = 15 * 60 * 1000;
+const MAX_JOB_LIFETIME_MS = 45 * 60 * 1000;
 
 // Dynamic concurrency control for video processing.
 //
@@ -268,15 +270,44 @@ export function cleanupExpiredJobs(): number {
 	let cleaned = 0;
 
 	for (const [jobId, job] of jobs) {
-		if (now - job.updatedAt > JOB_TTL_MS) {
+		const age = now - job.createdAt;
+		const staleness = now - job.updatedAt;
+
+		if (staleness > 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-manager] Cleaning up expired job ${jobId} (phase=${job.phase}, age=${Math.round(age / 60000)}m)`,
 				);
 				job.abortController?.abort();
 			}
 			deleteJob(jobId);
 			cleaned++;
+			continue;
+		}
+
+		if (isActivePhase(job.phase) && staleness > STALE_JOB_MS) {
+			console.warn(
+				`[job-manager] Marking stale job ${jobId} as error (phase=${job.phase}, no update for ${Math.round(staleness / 60000)}m)`,
+			);
+			job.abortController?.abort();
+			job.phase = "error";
+			job.error = `Job stale: no progress update for ${Math.round(staleness / 60000)} minutes`;
+			job.message = "Processing failed (stale)";
+			job.updatedAt = now;
+			cleaned++;
+			continue;
+		}
+
+		if (isActivePhase(job.phase) && age > MAX_JOB_LIFETIME_MS) {
+			console.warn(
+				`[job-manager] Marking long-running job ${jobId} as error (phase=${job.phase}, age=${Math.round(age / 60000)}m)`,
+			);
+			job.abortController?.abort();
+			job.phase = "error";
+			job.error = `Job exceeded maximum lifetime of ${Math.round(MAX_JOB_LIFETIME_MS / 60000)} minutes`;
+			job.message = "Processing failed (timeout)";
+			job.updatedAt = now;
+			cleaned++;
 		}
 	}
 
@@ -315,14 +346,32 @@ export async function sendWebhook(job: Job): Promise<void> {
 	}
 }
 
-const cleanupInterval = setInterval(
-	() => {
-		const cleaned = cleanupExpiredJobs();
-		if (cleaned > 0) {
-			console.log(`[job-manager] Cleaned up ${cleaned} expired jobs`);
+export function forceCleanupActiveJobs(): number {
+	let cleaned = 0;
+	const now = Date.now();
+
+	for (const [jobId, job] of jobs) {
+		if (isActivePhase(job.phase)) {
+			console.warn(
+				`[job-manager] Force-cleaning job ${jobId} (phase=${job.phase}, age=${Math.round((now - job.createdAt) / 60000)}m)`,
+			);
+			job.abortController?.abort();
+			job.phase = "error";
+			job.error = "Force-cleaned by admin";
+			job.message = "Processing failed (force-cleaned)";
+			job.updatedAt = now;
+			cleaned++;
 		}
-	},
-	5 * 60 * 1000,
-);
+	}
+
+	return cleaned;
+}
+
+const cleanupInterval = setInterval(() => {
+	const cleaned = cleanupExpiredJobs();
+	if (cleaned > 0) {
+		console.log(`[job-manager] Cleaned up ${cleaned} expired/stale jobs`);
+	}
+}, 60 * 1000);
 
 cleanupInterval.unref?.();

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

@@ -20,6 +20,7 @@ import {
 	canAcceptNewVideoProcess,
 	createJob,
 	deleteJob,
+	forceCleanupActiveJobs,
 	generateJobId,
 	getActiveVideoProcessCount,
 	getAllJobs,
@@ -77,6 +78,7 @@ function isTimeoutError(err: unknown): boolean {
 video.get("/status", (c) => {
 	const jobs = getAllJobs();
 	const resources = getSystemResources();
+	const now = Date.now();
 	return c.json({
 		instanceId: getInstanceId(),
 		pid: process.pid,
@@ -95,6 +97,9 @@ video.get("/status", (c) => {
 			progress: j.progress,
 			createdAt: j.createdAt,
 			updatedAt: j.updatedAt,
+			ageMinutes: Math.round((now - j.createdAt) / 60000),
+			stalenessMinutes: Math.round((now - j.updatedAt) / 60000),
+			error: j.error,
 		})),
 	});
 });
@@ -292,6 +297,19 @@ video.post("/process", async (c) => {
 			`[video/process] Async processing error for job ${jobId}:`,
 			err,
 		);
+		const currentJob = getJob(jobId);
+		if (
+			currentJob &&
+			currentJob.phase !== "error" &&
+			currentJob.phase !== "complete" &&
+			currentJob.phase !== "cancelled"
+		) {
+			updateJob(jobId, {
+				phase: "error",
+				error: err instanceof Error ? err.message : String(err),
+				message: "Processing failed (unhandled)",
+			});
+		}
 	});
 
 	return c.json({
@@ -754,4 +772,13 @@ video.post("/cleanup", async (c) => {
 	});
 });
 
+video.post("/force-cleanup", (c) => {
+	const cleaned = forceCleanupActiveJobs();
+	return c.json({
+		success: true,
+		cleanedJobs: cleaned,
+		message: `Force-cleaned ${cleaned} active jobs`,
+	});
+});
+
 export default video;