From 0988c6bd84a3cbc08ff8a054f2e2511060076383 Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Wed, 15 Jul 2026 17:51:03 +0300 Subject: [PATCH 01/17] fix: use container-aware media admission --- .../src/__tests__/lib/container-cpu.test.ts | 52 ++++++++ .../__tests__/lib/container-memory.test.ts | 53 ++++++++ .../src/__tests__/routes/health.test.ts | 4 + apps/media-server/src/lib/container-cpu.ts | 68 ++++++++++ apps/media-server/src/lib/container-memory.ts | 70 +++++++++++ apps/media-server/src/lib/job-manager.ts | 118 +++++++++++------- apps/media-server/src/lib/media-operations.ts | 11 +- apps/media-server/src/routes/health.ts | 16 +-- 8 files changed, 334 insertions(+), 58 deletions(-) create mode 100644 apps/media-server/src/__tests__/lib/container-cpu.test.ts create mode 100644 apps/media-server/src/__tests__/lib/container-memory.test.ts create mode 100644 apps/media-server/src/lib/container-cpu.ts create mode 100644 apps/media-server/src/lib/container-memory.ts diff --git a/apps/media-server/src/__tests__/lib/container-cpu.test.ts b/apps/media-server/src/__tests__/lib/container-cpu.test.ts new file mode 100644 index 00000000000..8bb76ca386b --- /dev/null +++ b/apps/media-server/src/__tests__/lib/container-cpu.test.ts @@ -0,0 +1,52 @@ +import { afterEach, describe, expect, test } from "bun:test"; +import { mkdtemp, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { + getContainerCpuLimit, + getContainerCpuUsageMicros, +} from "../../lib/container-cpu"; + +const tempDirs: string[] = []; + +afterEach(async () => { + await Promise.all( + tempDirs + .splice(0) + .map((path) => rm(path, { recursive: true, force: true })), + ); +}); + +describe("container CPU metrics", () => { + test("reads a cgroup v2 CPU quota", async () => { + const dir = await mkdtemp(join(tmpdir(), "cap-container-cpu-")); + tempDirs.push(dir); + const cpuMaxPath = join(dir, "cpu.max"); + await writeFile(cpuMaxPath, "200000 100000"); + + expect( + getContainerCpuLimit({ + cpuMaxPath, + cpuQuotaPath: join(dir, "missing-quota"), + cpuPeriodPath: join(dir, "missing-period"), + }), + ).toBe(2); + }); + + test("reads cgroup v2 aggregate CPU usage", async () => { + const dir = await mkdtemp(join(tmpdir(), "cap-container-cpu-")); + tempDirs.push(dir); + const cpuStatPath = join(dir, "cpu.stat"); + await writeFile( + cpuStatPath, + "usage_usec 1234567\nuser_usec 1000000\nsystem_usec 234567\n", + ); + + expect( + getContainerCpuUsageMicros({ + cpuStatPath, + cpuUsagePath: join(dir, "missing-usage"), + }), + ).toBe(1234567); + }); +}); diff --git a/apps/media-server/src/__tests__/lib/container-memory.test.ts b/apps/media-server/src/__tests__/lib/container-memory.test.ts new file mode 100644 index 00000000000..71e215e80be --- /dev/null +++ b/apps/media-server/src/__tests__/lib/container-memory.test.ts @@ -0,0 +1,53 @@ +import { afterEach, describe, expect, test } from "bun:test"; +import { mkdtemp, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { getContainerMemoryMetrics } from "../../lib/container-memory"; + +const tempDirs: string[] = []; + +afterEach(async () => { + await Promise.all( + tempDirs + .splice(0) + .map((path) => rm(path, { recursive: true, force: true })), + ); +}); + +describe("container memory metrics", () => { + test("reads cgroup usage and limit values", async () => { + const dir = await mkdtemp(join(tmpdir(), "cap-container-memory-")); + tempDirs.push(dir); + const limitPath = join(dir, "memory.max"); + const usagePath = join(dir, "memory.current"); + await writeFile(limitPath, String(1024 * 1024 * 1000)); + await writeFile(usagePath, String(1024 * 1024 * 750)); + + const metrics = getContainerMemoryMetrics({ + limitPaths: [limitPath], + usagePaths: [usagePath], + configuredLimitMB: 0, + }); + + expect(metrics.limitMB).toBe(1000); + expect(metrics.usageMB).toBe(750); + expect(metrics.pressure).toBe(0.75); + }); + + test("prefers an explicit limit while retaining cgroup usage", async () => { + const dir = await mkdtemp(join(tmpdir(), "cap-container-memory-")); + tempDirs.push(dir); + const usagePath = join(dir, "memory.current"); + await writeFile(usagePath, String(1024 * 1024 * 900)); + + const metrics = getContainerMemoryMetrics({ + limitPaths: [], + usagePaths: [usagePath], + configuredLimitMB: 1200, + }); + + expect(metrics.limitMB).toBe(1200); + expect(metrics.usageMB).toBe(900); + expect(metrics.pressure).toBe(0.75); + }); +}); diff --git a/apps/media-server/src/__tests__/routes/health.test.ts b/apps/media-server/src/__tests__/routes/health.test.ts index 68552268705..5d6d04e808d 100644 --- a/apps/media-server/src/__tests__/routes/health.test.ts +++ b/apps/media-server/src/__tests__/routes/health.test.ts @@ -12,5 +12,9 @@ describe("GET /health", () => { expect(data.mediaEngine.available).toBe(true); expect(typeof data.mediaEngine.version).toBe("string"); expect(data["ff" + "mpeg"]).toBeDefined(); + expect(typeof data.system.containerMemoryUsageMB).toBe("number"); + expect(typeof data.system.containerMemoryLimitMB).toBe("number"); + expect(typeof data.system.memoryPressure).toBe("number"); + expect(typeof data.system.processRssMB).toBe("number"); }); }); diff --git a/apps/media-server/src/lib/container-cpu.ts b/apps/media-server/src/lib/container-cpu.ts new file mode 100644 index 00000000000..95644d13ecb --- /dev/null +++ b/apps/media-server/src/lib/container-cpu.ts @@ -0,0 +1,68 @@ +import { existsSync, readFileSync } from "node:fs"; + +const CGROUP_V2_CPU_MAX_PATH = "/sys/fs/cgroup/cpu.max"; +const CGROUP_V1_CPU_QUOTA_PATH = "/sys/fs/cgroup/cpu/cpu.cfs_quota_us"; +const CGROUP_V1_CPU_PERIOD_PATH = "/sys/fs/cgroup/cpu/cpu.cfs_period_us"; +const CGROUP_V2_CPU_STAT_PATH = "/sys/fs/cgroup/cpu.stat"; +const CGROUP_V1_CPU_USAGE_PATH = "/sys/fs/cgroup/cpuacct/cpuacct.usage"; + +function readValue(path: string): string | null { + if (!existsSync(path)) return null; + try { + return readFileSync(path, "utf8").trim() || null; + } catch { + return null; + } +} + +export interface ContainerCpuOptions { + cpuMaxPath?: string; + cpuQuotaPath?: string; + cpuPeriodPath?: string; + cpuStatPath?: string; + cpuUsagePath?: string; +} + +export function getContainerCpuLimit( + options: ContainerCpuOptions = {}, +): number { + const cpuMax = readValue(options.cpuMaxPath ?? CGROUP_V2_CPU_MAX_PATH); + if (cpuMax) { + const [quotaValue, periodValue] = cpuMax.split(/\s+/); + if (quotaValue !== "max") { + const quota = Number.parseInt(quotaValue ?? "0", 10); + const period = Number.parseInt(periodValue ?? "0", 10); + if (quota > 0 && period > 0) return quota / period; + } + } + + const quota = Number.parseInt( + readValue(options.cpuQuotaPath ?? CGROUP_V1_CPU_QUOTA_PATH) ?? "0", + 10, + ); + const period = Number.parseInt( + readValue(options.cpuPeriodPath ?? CGROUP_V1_CPU_PERIOD_PATH) ?? "0", + 10, + ); + return quota > 0 && period > 0 ? quota / period : 0; +} + +export function getContainerCpuUsageMicros( + options: ContainerCpuOptions = {}, +): number { + const cpuStat = readValue(options.cpuStatPath ?? CGROUP_V2_CPU_STAT_PATH); + if (cpuStat) { + for (const line of cpuStat.split(/\r?\n/)) { + const [name, value] = line.trim().split(/\s+/); + if (name !== "usage_usec") continue; + const usage = Number.parseInt(value ?? "0", 10); + if (usage >= 0) return usage; + } + } + + const usageNanos = Number.parseInt( + readValue(options.cpuUsagePath ?? CGROUP_V1_CPU_USAGE_PATH) ?? "0", + 10, + ); + return usageNanos > 0 ? usageNanos / 1000 : 0; +} diff --git a/apps/media-server/src/lib/container-memory.ts b/apps/media-server/src/lib/container-memory.ts new file mode 100644 index 00000000000..22633dd2cca --- /dev/null +++ b/apps/media-server/src/lib/container-memory.ts @@ -0,0 +1,70 @@ +import { existsSync, readFileSync } from "node:fs"; + +const CGROUP_MEMORY_LIMIT_PATHS = [ + "/sys/fs/cgroup/memory.max", + "/sys/fs/cgroup/memory/memory.limit_in_bytes", +]; +const CGROUP_MEMORY_USAGE_PATHS = [ + "/sys/fs/cgroup/memory.current", + "/sys/fs/cgroup/memory/memory.usage_in_bytes", +]; +const MAX_PLAUSIBLE_CONTAINER_LIMIT_BYTES = 1024 ** 5; + +function readMemoryValueMB(paths: string[], enforcePlausibleLimit: boolean) { + for (const path of paths) { + if (!existsSync(path)) continue; + + let rawValue: string; + try { + rawValue = readFileSync(path, "utf8").trim(); + } catch { + continue; + } + + if (!rawValue || rawValue === "max") continue; + + const bytes = Number.parseInt(rawValue, 10); + if ( + Number.isFinite(bytes) && + bytes > 0 && + (!enforcePlausibleLimit || bytes < MAX_PLAUSIBLE_CONTAINER_LIMIT_BYTES) + ) { + return Math.round(bytes / (1024 * 1024)); + } + } + + return 0; +} + +export interface ContainerMemoryMetrics { + usageMB: number; + limitMB: number; + pressure: number; +} + +interface ContainerMemoryOptions { + limitPaths?: string[]; + usagePaths?: string[]; + configuredLimitMB?: number; +} + +export function getContainerMemoryMetrics( + options: ContainerMemoryOptions = {}, +): ContainerMemoryMetrics { + const configuredLimitMB = + options.configuredLimitMB ?? + (Number.parseInt(process.env.MEDIA_SERVER_MEMORY_LIMIT_MB ?? "0", 10) || 0); + const limitMB = + configuredLimitMB || + readMemoryValueMB(options.limitPaths ?? CGROUP_MEMORY_LIMIT_PATHS, true); + const usageMB = readMemoryValueMB( + options.usagePaths ?? CGROUP_MEMORY_USAGE_PATHS, + false, + ); + + return { + usageMB, + limitMB, + pressure: limitMB > 0 && usageMB > 0 ? usageMB / limitMB : 0, + }; +} diff --git a/apps/media-server/src/lib/job-manager.ts b/apps/media-server/src/lib/job-manager.ts index 8ef8132dad5..8c6c513c864 100644 --- a/apps/media-server/src/lib/job-manager.ts +++ b/apps/media-server/src/lib/job-manager.ts @@ -1,5 +1,9 @@ -import { existsSync, readFileSync } from "node:fs"; import os from "node:os"; +import { + getContainerCpuLimit, + getContainerCpuUsageMicros, +} from "./container-cpu"; +import { getContainerMemoryMetrics } from "./container-memory"; import type { MediaOperationHandle } from "./media-operations"; import type { TempFileHandle } from "./temp-files"; import { getActiveDirectVideoProcessCount } from "./video-capacity"; @@ -73,49 +77,54 @@ const configuredMaxProcesses = 10, ) || 0; -const cpuCount = os.cpus().length; +const hostCpuCount = os.cpus().length; const CPU_LOAD_THRESHOLD = 0.8; +const CPU_REJECT_THRESHOLD = 0.95; const DEFAULT_MAX_CONCURRENT_VIDEO_PROCESSES = 4; -const CGROUP_MEMORY_LIMIT_PATHS = [ - "/sys/fs/cgroup/memory.max", - "/sys/fs/cgroup/memory/memory.limit_in_bytes", -]; -const MAX_PLAUSIBLE_CONTAINER_LIMIT_BYTES = 1024 ** 5; const MEMORY_THROTTLE_THRESHOLD = 0.85; -const MEMORY_REJECT_THRESHOLD = 0.95; +const MEMORY_REJECT_THRESHOLD = 0.9; const VIDEO_PROCESS_MEMORY_BUDGET_MB = 768; +const CPU_SAMPLE_MIN_INTERVAL_MS = 250; -function readContainerMemoryLimitMB(): number { - for (const path of CGROUP_MEMORY_LIMIT_PATHS) { - if (!existsSync(path)) continue; +let previousContainerCpuUsageMicros = 0; +let previousContainerCpuSampleAt = 0; +let containerCpuPressure = 0; - let rawValue: string; - try { - rawValue = readFileSync(path, "utf8").trim(); - } catch { - continue; - } +function getCpuCapacity(): number { + return getContainerCpuLimit() || hostCpuCount; +} - if (!rawValue || rawValue === "max") continue; +function getCpuPressure(cpuCapacity: number, loadAvg1m: number): number { + const cpuLimit = getContainerCpuLimit(); + const usageMicros = getContainerCpuUsageMicros(); + const now = performance.now(); - const bytes = Number.parseInt(rawValue, 10); + if (usageMicros > 0 && cpuLimit > 0) { if ( - Number.isFinite(bytes) && - bytes > 0 && - bytes < MAX_PLAUSIBLE_CONTAINER_LIMIT_BYTES + previousContainerCpuUsageMicros > 0 && + now - previousContainerCpuSampleAt >= CPU_SAMPLE_MIN_INTERVAL_MS ) { - return Math.floor(bytes / (1024 * 1024)); + const elapsedSeconds = (now - previousContainerCpuSampleAt) / 1000; + const usedCpuSeconds = + (usageMicros - previousContainerCpuUsageMicros) / 1_000_000; + containerCpuPressure = + usedCpuSeconds >= 0 + ? Math.max(0, usedCpuSeconds / elapsedSeconds / cpuCapacity) + : 0; + previousContainerCpuUsageMicros = usageMicros; + previousContainerCpuSampleAt = now; + } else if (previousContainerCpuUsageMicros === 0) { + previousContainerCpuUsageMicros = usageMicros; + previousContainerCpuSampleAt = now; } + + return containerCpuPressure; } - return 0; + return loadAvg1m / cpuCapacity; } -const PROCESS_RSS_LIMIT_MB = - Number.parseInt(process.env.MEDIA_SERVER_MEMORY_LIMIT_MB ?? "0", 10) || - readContainerMemoryLimitMB(); - function isActivePhase(phase: JobPhase): boolean { return phase !== "complete" && phase !== "error" && phase !== "cancelled"; } @@ -134,12 +143,13 @@ export function getMaxConcurrentVideoProcesses(): number { if (configuredMaxProcesses > 0) { return configuredMaxProcesses; } + const containerMemoryLimitMB = getContainerMemoryMetrics().limitMB; const memoryBoundMax = - PROCESS_RSS_LIMIT_MB > 0 + containerMemoryLimitMB > 0 ? Math.max( 1, Math.floor( - (PROCESS_RSS_LIMIT_MB * MEMORY_THROTTLE_THRESHOLD) / + (containerMemoryLimitMB * MEMORY_THROTTLE_THRESHOLD) / VIDEO_PROCESS_MEMORY_BUDGET_MB, ), ) @@ -148,7 +158,7 @@ export function getMaxConcurrentVideoProcesses(): number { 1, Math.min( DEFAULT_MAX_CONCURRENT_VIDEO_PROCESSES, - Math.floor(cpuCount / 2), + Math.floor(getCpuCapacity() / 2), memoryBoundMax, ), ); @@ -156,11 +166,15 @@ export function getMaxConcurrentVideoProcesses(): number { export interface SystemResources { cpuCount: number; + hostCpuCount: number; loadAvg1m: number; cpuPressure: number; processRssMB: number; processHeapMB: number; processRssLimitMB: number; + containerMemoryUsageMB: number; + containerMemoryLimitMB: number; + memoryPressure: number; configuredMax: number; effectiveMax: number; throttleReason: string | null; @@ -168,51 +182,63 @@ export interface SystemResources { export function getSystemResources(): SystemResources { const loadAvg1m = os.loadavg()[0]; - const cpuPressure = loadAvg1m / cpuCount; + const cpuCount = getCpuCapacity(); + const cpuPressure = getCpuPressure(cpuCount, loadAvg1m); const mem = process.memoryUsage(); const processRssMB = Math.round(mem.rss / (1024 * 1024)); const processHeapMB = Math.round(mem.heapUsed / (1024 * 1024)); + const containerMemory = getContainerMemoryMetrics(); + const memoryUsageMB = containerMemory.usageMB || processRssMB; + const memoryLimitMB = containerMemory.limitMB; + const memoryPressure = memoryLimitMB > 0 ? memoryUsageMB / memoryLimitMB : 0; const max = getMaxConcurrentVideoProcesses(); let effectiveMax = max; let throttleReason: string | null = null; 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`; + effectiveMax = + cpuPressure >= CPU_REJECT_THRESHOLD + ? 0 + : Math.max( + 1, + Math.floor(max * (1 - (cpuPressure - CPU_LOAD_THRESHOLD))), + ); + throttleReason = `CPU utilization ${cpuPressure.toFixed(2)} exceeds ${CPU_LOAD_THRESHOLD} threshold`; } - if ( - PROCESS_RSS_LIMIT_MB > 0 && - processRssMB > PROCESS_RSS_LIMIT_MB * MEMORY_THROTTLE_THRESHOLD - ) { - const memPressure = processRssMB / PROCESS_RSS_LIMIT_MB; + if (memoryPressure > MEMORY_THROTTLE_THRESHOLD) { const memMax = - memPressure >= MEMORY_REJECT_THRESHOLD + memoryPressure >= MEMORY_REJECT_THRESHOLD ? 0 - : Math.max(1, Math.floor(max * (1 - memPressure))); + : Math.max(1, Math.floor(max * (1 - memoryPressure))); if (memMax < effectiveMax) { effectiveMax = memMax; - throttleReason = `Process RSS ${processRssMB}MB exceeds ${Math.round(MEMORY_THROTTLE_THRESHOLD * 100)}% of ${PROCESS_RSS_LIMIT_MB}MB limit`; + throttleReason = `Container memory ${memoryUsageMB}MB exceeds ${Math.round(MEMORY_THROTTLE_THRESHOLD * 100)}% of ${memoryLimitMB}MB limit`; } } return { cpuCount, + hostCpuCount, loadAvg1m, cpuPressure, processRssMB, processHeapMB, - processRssLimitMB: PROCESS_RSS_LIMIT_MB, + processRssLimitMB: memoryLimitMB, + containerMemoryUsageMB: containerMemory.usageMB, + containerMemoryLimitMB: containerMemory.limitMB, + memoryPressure, configuredMax: configuredMaxProcesses, effectiveMax, throttleReason, }; } +export function hasCriticalMemoryPressure(): boolean { + return getSystemResources().memoryPressure >= MEMORY_REJECT_THRESHOLD; +} + export function canAcceptNewVideoProcess(): boolean { const active = getActiveVideoProcessCount(); const resources = getSystemResources(); diff --git a/apps/media-server/src/lib/media-operations.ts b/apps/media-server/src/lib/media-operations.ts index 53f17fa5966..1c9901dd89e 100644 --- a/apps/media-server/src/lib/media-operations.ts +++ b/apps/media-server/src/lib/media-operations.ts @@ -1,3 +1,4 @@ +import { hasCriticalMemoryPressure } from "./job-manager"; import { terminateAllSubprocesses } from "./subprocess"; export interface MediaOperationHandle { @@ -30,7 +31,10 @@ export function getActiveAudioOperationCount(): number { } export function canAcceptNewAudioOperation(): boolean { - return getActiveAudioOperationCount() < MAX_CONCURRENT_AUDIO_OPERATIONS; + return ( + !hasCriticalMemoryPressure() && + getActiveAudioOperationCount() < MAX_CONCURRENT_AUDIO_OPERATIONS + ); } export function getActiveProbeOperationCount(): number { @@ -42,7 +46,10 @@ export function getActiveProbeOperationCount(): number { } export function canAcceptNewProbeOperation(): boolean { - return getActiveProbeOperationCount() < MAX_CONCURRENT_PROBE_OPERATIONS; + return ( + !hasCriticalMemoryPressure() && + getActiveProbeOperationCount() < MAX_CONCURRENT_PROBE_OPERATIONS + ); } export function registerMediaOperation( diff --git a/apps/media-server/src/routes/health.ts b/apps/media-server/src/routes/health.ts index 9501f32d5b5..f8a640af688 100644 --- a/apps/media-server/src/routes/health.ts +++ b/apps/media-server/src/routes/health.ts @@ -1,33 +1,29 @@ import os from "node:os"; import { Hono } from "hono"; +import { getSystemResources } from "../lib/job-manager"; import { getMediaEngineStatus } from "../lib/media-engine"; const health = new Hono(); health.get("/", (c) => { const mediaEngine = getMediaEngineStatus(); - const cpuCount = os.cpus().length; const loadAvg = os.loadavg(); - const totalMemMB = Math.round(os.totalmem() / (1024 * 1024)); - const freeMemMB = Math.round(os.freemem() / (1024 * 1024)); - const memoryUsagePercent = 1 - os.freemem() / os.totalmem(); + const resources = getSystemResources(); return c.json({ - status: mediaEngine.available ? "ok" : "degraded", + status: + mediaEngine.available && resources.effectiveMax > 0 ? "ok" : "degraded", mediaEngine, ["ff" + "mpeg"]: { available: mediaEngine.available, version: mediaEngine.version, }, system: { - cpuCount, + ...resources, loadAvg1m: loadAvg[0], loadAvg5m: loadAvg[1], loadAvg15m: loadAvg[2], - totalMemoryMB: totalMemMB, - freeMemoryMB: freeMemMB, - memoryUsagePercent: Math.round(memoryUsagePercent * 100), - uptimeSeconds: Math.round(os.uptime()), + uptimeSeconds: Math.round(process.uptime()), }, }); }); From 3e4e444e8ad3a79b4d3321d1bb4732fb0ddd7b0d Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Wed, 15 Jul 2026 17:51:09 +0300 Subject: [PATCH 02/17] fix: harden streaming media inputs --- .../lib/media-audio.integration.test.ts | 84 ++++++++ apps/media-server/src/lib/media-audio.ts | 182 ++++++++++++++++-- apps/media-server/src/lib/media-video.ts | 40 +++- 3 files changed, 276 insertions(+), 30 deletions(-) diff --git a/apps/media-server/src/__tests__/lib/media-audio.integration.test.ts b/apps/media-server/src/__tests__/lib/media-audio.integration.test.ts index 4101df81339..30608519ce7 100644 --- a/apps/media-server/src/__tests__/lib/media-audio.integration.test.ts +++ b/apps/media-server/src/__tests__/lib/media-audio.integration.test.ts @@ -1,4 +1,6 @@ import { describe, expect, test } from "bun:test"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; import { join } from "node:path"; import { checkHasAudioTrack, @@ -11,6 +13,41 @@ const FIXTURES_DIR = join(import.meta.dir, "..", "fixtures"); const TEST_VIDEO_WITH_AUDIO = `file://${join(FIXTURES_DIR, "test-with-audio.mp4")}`; const TEST_VIDEO_NO_AUDIO = `file://${join(FIXTURES_DIR, "test-no-audio.mp4")}`; +async function createHlsFixture(): Promise { + const dirPath = await mkdtemp(join(tmpdir(), "cap-audio-hls-")); + const proc = Bun.spawn({ + cmd: [ + "ffmpeg", + "-hide_banner", + "-loglevel", + "error", + "-y", + "-i", + join(FIXTURES_DIR, "test-with-audio.mp4"), + "-c", + "copy", + "-hls_time", + "0.25", + "-hls_list_size", + "0", + "-hls_segment_filename", + join(dirPath, "segment-%03d.ts"), + join(dirPath, "manifest.m3u8"), + ], + stdout: "ignore", + stderr: "pipe", + }); + const [stderr, exitCode] = await Promise.all([ + new Response(proc.stderr).text(), + proc.exited, + ]); + if (exitCode !== 0) { + await rm(dirPath, { recursive: true, force: true }); + throw new Error(`Failed to create HLS fixture: ${stderr}`); + } + return dirPath; +} + async function readStream(stream: ReadableStream) { const reader = stream.getReader(); const chunks: Uint8Array[] = []; @@ -66,6 +103,53 @@ describe("mediaAudio integration tests", () => { const hasAudio = await checkHasAudioTrack(TEST_VIDEO_NO_AUDIO); expect(hasAudio).toBe(false); }); + + test("inherits signed queries for relative HLS segments", async () => { + const dirPath = await createHlsFixture(); + const requests: string[] = []; + const server = Bun.serve({ + port: 0, + fetch(request) { + const url = new URL(request.url); + requests.push(`${url.pathname}${url.search}`); + if ( + url.pathname !== "/manifest.m3u8" && + url.searchParams.get("token") !== "signed" + ) { + return new Response("Forbidden", { status: 403 }); + } + return new Response(Bun.file(join(dirPath, url.pathname.slice(1)))); + }, + }); + + try { + const hasAudio = await checkHasAudioTrack( + `${server.url}manifest.m3u8?token=signed`, + ); + expect(hasAudio).toBe(true); + expect(requests).toContain("/segment-000.ts?token=signed"); + } finally { + await server.stop(true); + await rm(dirPath, { recursive: true, force: true }); + } + }); + + test("contains upstream failures without leaking the active operation", async () => { + const beforeCount = getActiveProcessCount(); + const server = Bun.serve({ + port: 0, + fetch: () => new Response("Unavailable", { status: 503 }), + }); + + try { + await expect( + checkHasAudioTrack(`${server.url}video.mp4`), + ).rejects.toThrow("FFprobe exited"); + await waitForAudioOperations(beforeCount); + } finally { + await server.stop(true); + } + }); }); describe("extractAudio", () => { diff --git a/apps/media-server/src/lib/media-audio.ts b/apps/media-server/src/lib/media-audio.ts index 67c6948e458..9f88c5e5032 100644 --- a/apps/media-server/src/lib/media-audio.ts +++ b/apps/media-server/src/lib/media-audio.ts @@ -1,5 +1,7 @@ +import { mkdtemp, rm } from "node:fs/promises"; +import { join } from "node:path"; import { type Subprocess, spawn } from "bun"; -import { createMediaInput, withTimeout } from "./media-common"; +import { withTimeout } from "./media-common"; import { canAcceptNewAudioOperation, getActiveAudioOperationCount, @@ -7,8 +9,9 @@ import { unregisterMediaOperation, withMediaOperation, } from "./media-operations"; -import { checkVideoAccessible } from "./media-probe"; +import { materializeStreamingInput } from "./media-video"; import { registerSubprocess, terminateProcess } from "./subprocess"; +import { ensureTempDir, getTempDir } from "./temp-files"; export interface AudioExtractionOptions { format?: "mp3"; @@ -26,6 +29,8 @@ const CHECK_TIMEOUT_MS = 30_000; const EXTRACT_TIMEOUT_MS = 120_000; const MAX_AUDIO_SIZE_BYTES = 100 * 1024 * 1024; const MAX_STDERR_BYTES = 64 * 1024; +const AUDIO_PROBE_MAX_ATTEMPTS = 3; +const AUDIO_PROBE_RETRY_BASE_MS = 250; const DEFAULT_OPTIONS: Required = { format: "mp3", @@ -98,7 +103,10 @@ function redactUrl(value: string): string { } function redactProcessOutput(output: string, url: string): string { - return output.split(url).join(redactUrl(url)); + return output + .split(url) + .join(redactUrl(url)) + .replace(/https?:\/\/[^\s"'<>]+/g, redactUrl); } function getAudioExtractArgs( @@ -120,34 +128,168 @@ function getAudioExtractArgs( ]; } +function getAudioProbeArgs(inputPath: string): string[] { + const normalizedPath = (inputPath.split("?")[0] ?? "").toLowerCase(); + const args = ["ffprobe", "-v", "error"]; + if (normalizedPath.endsWith(".m3u8") || normalizedPath.endsWith(".mpd")) { + args.push("-protocol_whitelist", "file,http,https,tcp,tls,crypto,data"); + } + if (normalizedPath.endsWith(".m3u8")) { + args.push( + "-allowed_extensions", + "ALL", + "-allowed_segment_extensions", + "ALL", + "-extension_picky", + "0", + ); + } + args.push("-show_entries", "stream=codec_type", "-of", "csv=p=0", inputPath); + return args; +} + +function isRemoteStreamingUrl(value: string): boolean { + try { + const url = new URL(value); + if (url.protocol !== "http:" && url.protocol !== "https:") return false; + const pathname = url.pathname.toLowerCase(); + return pathname.endsWith(".m3u8") || pathname.endsWith(".mpd"); + } catch { + return false; + } +} + +async function prepareAudioProbeInput( + videoUrl: string, + abortSignal: AbortSignal, +): Promise<{ inputPath: string; cleanup: () => Promise }> { + if (!isRemoteStreamingUrl(videoUrl)) { + return { inputPath: videoUrl, cleanup: async () => {} }; + } + + await ensureTempDir(); + const dirPath = await mkdtemp(join(getTempDir(), "audio-probe-")); + try { + const inputPath = await materializeStreamingInput( + videoUrl, + dirPath, + abortSignal, + ); + return { + inputPath, + cleanup: async () => { + await rm(dirPath, { recursive: true, force: true }); + }, + }; + } catch (error) { + await rm(dirPath, { recursive: true, force: true }).catch(() => {}); + throw error; + } +} + +async function probeAudioTracks( + inputPath: string, + sourceUrl: string, + setCurrentCancel: (cancel: () => Promise | void) => void, + isCancelled: () => boolean, +): Promise { + let lastError: Error | undefined; + + for (let attempt = 0; attempt < AUDIO_PROBE_MAX_ATTEMPTS; attempt++) { + if (isCancelled()) { + throw new Error("Audio probe cancelled"); + } + + const proc = registerSubprocess( + spawn({ + cmd: getAudioProbeArgs(inputPath), + stdout: "pipe", + stderr: "pipe", + }), + ); + setCurrentCancel(() => terminateProcess(proc)); + + try { + const [stdoutText, stderrText, exitCode] = await Promise.all([ + readStreamWithLimit( + proc.stdout as ReadableStream, + MAX_STDERR_BYTES, + ), + readStreamWithLimit( + proc.stderr as ReadableStream, + MAX_STDERR_BYTES, + ), + proc.exited, + ]); + const safeStderrText = redactProcessOutput(stderrText, sourceUrl); + + if (exitCode === 0) { + const trackTypes = stdoutText + .split(/\r?\n/) + .map((value) => value.trim()) + .filter(Boolean); + if (!trackTypes.includes("video")) { + throw new Error("No video stream found"); + } + return trackTypes.includes("audio"); + } + + lastError = new Error( + `FFprobe exited with code ${exitCode}: ${safeStderrText}`, + ); + } finally { + await terminateProcess(proc); + } + + if (attempt < AUDIO_PROBE_MAX_ATTEMPTS - 1) { + await Bun.sleep(AUDIO_PROBE_RETRY_BASE_MS * 2 ** attempt); + } + } + + throw lastError ?? new Error("FFprobe could not inspect the video"); +} + export async function checkHasAudioTrack(videoUrl: string): Promise { if (!canAcceptNewAudioOperation()) { throw new Error("Server is busy, please try again later"); } - return await withMediaOperation("audio", () => - withTimeout( + return await withMediaOperation("audio", async (setCancel) => { + let cancelled = false; + let cancelCurrent: () => Promise | void = () => {}; + let cleanupInput: () => Promise = async () => {}; + const abortController = new AbortController(); + const cancel = async () => { + cancelled = true; + abortController.abort(); + await cancelCurrent(); + await cleanupInput(); + }; + setCancel(cancel); + return await withTimeout( (async () => { - const input = createMediaInput(videoUrl); + const preparedInput = await prepareAudioProbeInput( + videoUrl, + abortController.signal, + ); + cleanupInput = preparedInput.cleanup; try { - const videoTrack = await input.getPrimaryVideoTrack(); - if (!videoTrack) { - if (!(await checkVideoAccessible(videoUrl))) { - throw new Error( - "Media engine could not read video file: no streams detected", - ); - } - throw new Error("No video stream found"); - } - const audioTrack = await input.getPrimaryAudioTrack(); - return Boolean(audioTrack); + return await probeAudioTracks( + preparedInput.inputPath, + videoUrl, + (nextCancel) => { + cancelCurrent = nextCancel; + }, + () => cancelled, + ); } finally { - input.dispose(); + await cleanupInput(); } })(), CHECK_TIMEOUT_MS, - ), - ); + cancel, + ); + }); } export async function extractAudio( diff --git a/apps/media-server/src/lib/media-video.ts b/apps/media-server/src/lib/media-video.ts index 7b9fb8d9553..2552eccd8e9 100644 --- a/apps/media-server/src/lib/media-video.ts +++ b/apps/media-server/src/lib/media-video.ts @@ -190,6 +190,13 @@ function resolveResourceUrl( return withQuery(new URL(resource, baseUrl).toString(), query); } +function getFetchSignal(timeoutMs: number, abortSignal?: AbortSignal) { + const timeoutSignal = AbortSignal.timeout(timeoutMs); + return abortSignal + ? AbortSignal.any([abortSignal, timeoutSignal]) + : timeoutSignal; +} + function redactUrl(value: string): string { try { const url = new URL(value); @@ -227,12 +234,13 @@ export async function materializeHlsPlaylist( playlistUrl: string, dirPath: string, cache = new Map(), + abortSignal?: AbortSignal, ): Promise { const cached = cache.get(playlistUrl); if (cached) return cached; const response = await fetch(playlistUrl, { - signal: AbortSignal.timeout(DOWNLOAD_TIMEOUT_MS), + signal: getFetchSignal(DOWNLOAD_TIMEOUT_MS, abortSignal), }); if (!response.ok) { @@ -258,7 +266,7 @@ export async function materializeHlsPlaylist( if (!trimmed.startsWith("#")) { const resolved = resolveResourceUrl(trimmed, baseUrl, query); return isHlsUrl(resolved) - ? await materializeHlsPlaylist(resolved, dirPath, cache) + ? await materializeHlsPlaylist(resolved, dirPath, cache, abortSignal) : resolved; } @@ -273,7 +281,7 @@ export async function materializeHlsPlaylist( const resolved = resolveResourceUrl(original, baseUrl, query); const replacement = isHlsUrl(resolved) - ? await materializeHlsPlaylist(resolved, dirPath, cache) + ? await materializeHlsPlaylist(resolved, dirPath, cache, abortSignal) : resolved; rewritten = rewritten.replace( @@ -293,9 +301,10 @@ export async function materializeHlsPlaylist( export async function materializeMpdManifest( manifestUrl: string, dirPath: string, + abortSignal?: AbortSignal, ): Promise { const response = await fetch(manifestUrl, { - signal: AbortSignal.timeout(DOWNLOAD_TIMEOUT_MS), + signal: getFetchSignal(DOWNLOAD_TIMEOUT_MS, abortSignal), }); if (!response.ok) { @@ -656,9 +665,10 @@ function shouldFallbackToGenericMpd(error: unknown): boolean { export async function materializeMpdAsHlsPlaylist( manifestUrl: string, dirPath: string, + abortSignal?: AbortSignal, ): Promise { const response = await fetch(manifestUrl, { - signal: AbortSignal.timeout(DOWNLOAD_TIMEOUT_MS), + signal: getFetchSignal(DOWNLOAD_TIMEOUT_MS, abortSignal), }); if (!response.ok) { @@ -821,17 +831,23 @@ export async function materializeMpdAsHlsPlaylist( export async function materializeStreamingInput( videoUrl: string, dirPath: string, + abortSignal?: AbortSignal, ): Promise { if (isHlsUrl(videoUrl)) { - return await materializeHlsPlaylist(videoUrl, dirPath); + return await materializeHlsPlaylist( + videoUrl, + dirPath, + undefined, + abortSignal, + ); } if (isMpdUrl(videoUrl)) { try { - return await materializeMpdAsHlsPlaylist(videoUrl, dirPath); + return await materializeMpdAsHlsPlaylist(videoUrl, dirPath, abortSignal); } catch (err) { if (!shouldFallbackToGenericMpd(err)) throw err; - return await materializeMpdManifest(videoUrl, dirPath); + return await materializeMpdManifest(videoUrl, dirPath, abortSignal); } } @@ -996,7 +1012,11 @@ async function downloadStreamingVideoToTemp( }; try { - const inputPath = await materializeStreamingInput(videoUrl, manifestDir); + const inputPath = await materializeStreamingInput( + videoUrl, + manifestDir, + abortSignal, + ); await runFfmpegCommand( buildStreamingDownloadFfmpegArgs(inputPath, tempFile.path), @@ -1761,7 +1781,7 @@ async function uploadWithRetry( presignedUrl: string, contentType: string, contentLength: number, - bodyFactory: () => Blob | Uint8Array | ArrayBuffer | BunFile, + bodyFactory: () => Blob | BunFile, ): Promise { let lastError: Error | undefined; From fcd1078a1c1a1668a8604f9913d68699cae1f879 Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Wed, 15 Jul 2026 17:51:14 +0300 Subject: [PATCH 03/17] fix: stream large segment mux jobs --- ...edia-routes-real-world.integration.test.ts | 81 ++++- apps/media-server/src/routes/video.ts | 313 +++++++++++++----- 2 files changed, 310 insertions(+), 84 deletions(-) diff --git a/apps/media-server/src/__tests__/lib/media-routes-real-world.integration.test.ts b/apps/media-server/src/__tests__/lib/media-routes-real-world.integration.test.ts index 405f2d4c1eb..73d10ce2c67 100644 --- a/apps/media-server/src/__tests__/lib/media-routes-real-world.integration.test.ts +++ b/apps/media-server/src/__tests__/lib/media-routes-real-world.integration.test.ts @@ -33,6 +33,8 @@ let baseUrl = ""; let tempDir = ""; const uploadedArtifacts = new Map(); +let transientFixtureFailures = 0; +let permanentFixtureFailures = 0; function fileUrl(path: string) { return pathToFileURL(path).toString(); @@ -138,12 +140,26 @@ beforeAll(async () => { const url = new URL(request.url); if (request.method === "GET" || request.method === "HEAD") { + if (url.pathname === "/fixtures/permanent-unavailable.m4s") { + permanentFixtureFailures++; + return new Response("Unavailable", { status: 503 }); + } + if ( + request.method === "GET" && + url.pathname === "/fixtures/transient-no-audio.mp4" && + transientFixtureFailures < 2 + ) { + transientFixtureFailures++; + return new Response("Unavailable", { status: 503 }); + } const fixturePath = url.pathname === "/fixtures/test-no-audio.mp4" ? TEST_VIDEO_NO_AUDIO - : url.pathname === "/fixtures/test-with-audio.mp4" - ? TEST_VIDEO_WITH_AUDIO - : null; + : url.pathname === "/fixtures/transient-no-audio.mp4" + ? TEST_VIDEO_NO_AUDIO + : url.pathname === "/fixtures/test-with-audio.mp4" + ? TEST_VIDEO_WITH_AUDIO + : null; if (fixturePath) { const fixture = Bun.file(fixturePath); @@ -169,7 +185,7 @@ beforeAll(async () => { }; return request.method === "HEAD" ? new Response(null, { headers }) - : new Response(bytes, { headers }); + : new Response(Uint8Array.from(bytes).buffer, { headers }); } if (request.method === "PUT" && url.pathname.startsWith("/uploads/")) { @@ -196,6 +212,8 @@ beforeAll(async () => { beforeEach(() => { mock.restore(); uploadedArtifacts.clear(); + transientFixtureFailures = 0; + permanentFixtureFailures = 0; }); afterAll(() => { @@ -320,6 +338,61 @@ describe("media routes real-world integration tests", () => { } }, 90000); + test("retries transient segment downloads and completes a real mux job", async () => { + const response = await app.fetch( + mediaPostRequest("/video/mux-segments", { + videoId: "real-mux-video", + userId: "real-mux-user", + outputPresignedUrl: uploadUrl("mux-output.mp4"), + videoInitUrl: `${baseUrl}/fixtures/transient-no-audio.mp4`, + videoSegmentUrls: [], + }), + ); + + expect(response.status).toBe(200); + const data = (await response.json()) as { jobId: string }; + const job = await waitForTerminalJob(data.jobId); + try { + expect(job.phase).toBe("complete"); + expect(job.error).toBeUndefined(); + expect(transientFixtureFailures).toBe(2); + + const bytes = uploadedBytes("/uploads/mux-output.mp4"); + expectMp4(bytes); + const metadata = await probeBytesAsMp4(bytes, "mux-output.mp4"); + expect(metadata.videoCodec).toBe("h264"); + expect(metadata.audioCodec).toBeNull(); + } finally { + deleteJob(data.jobId); + } + }, 90000); + + test("fails a mux job when a segment stays unavailable after retries", async () => { + const response = await app.fetch( + mediaPostRequest("/video/mux-segments", { + videoId: "failed-segment-mux-video", + userId: "failed-segment-mux-user", + outputPresignedUrl: uploadUrl("failed-segment-output.mp4"), + videoInitUrl: fixtureUrl("test-no-audio.mp4"), + videoSegmentUrls: [`${baseUrl}/fixtures/permanent-unavailable.m4s`], + }), + ); + + expect(response.status).toBe(200); + const data = (await response.json()) as { jobId: string }; + const job = await waitForTerminalJob(data.jobId); + try { + expect(job.phase).toBe("error"); + expect(job.error).toContain("503"); + expect(permanentFixtureFailures).toBe(3); + expect(uploadedArtifacts.has("/uploads/failed-segment-output.mp4")).toBe( + false, + ); + } finally { + deleteJob(data.jobId); + } + }, 90000); + test("edits and uploads a real video job through the async route", async () => { const response = await app.fetch( mediaPostRequest("/video/edit", { diff --git a/apps/media-server/src/routes/video.ts b/apps/media-server/src/routes/video.ts index abc3158203e..4b4040c1cad 100644 --- a/apps/media-server/src/routes/video.ts +++ b/apps/media-server/src/routes/video.ts @@ -15,6 +15,7 @@ import { getJobProgress, getMaxConcurrentVideoProcesses, getSystemResources, + hasCriticalMemoryPressure, sendWebhook, touchJob, updateJob, @@ -56,6 +57,10 @@ const PROCESSING_HEARTBEAT_MS = 60 * 1000; const MEDIA_ENGINE_ERROR_CODE = ["FF", "MPEG_ERROR"].join(""); const PROBE_ERROR_CODE = ["FF", "PROBE_ERROR"].join(""); const VIDEO_BUSY_RETRY_AFTER_SECONDS = 15; +const VIDEO_MEMORY_PRESSURE_ERROR = + "SERVER_BUSY: container memory pressure is too high"; +const SEGMENT_DOWNLOAD_MAX_ATTEMPTS = 3; +const SEGMENT_DOWNLOAD_RETRY_BASE_MS = 250; const probeSchema = z.object({ videoUrl: z.string().url(), @@ -205,7 +210,11 @@ function summarizeVideoInput(videoUrl: string, inputExtension?: string) { } function isBusyError(err: unknown): boolean { - return err instanceof Error && err.message.includes("Server is busy"); + return ( + err instanceof Error && + (err.message.includes("Server is busy") || + err.message.includes("SERVER_BUSY")) + ); } function isTimeoutError(err: unknown): boolean { @@ -300,6 +309,34 @@ async function withJobHeartbeat( } } +async function withMuxMemoryGuard( + abortController: AbortController, + operation: () => Promise, +): Promise { + let memoryPressureExceeded = false; + const interval = setInterval(() => { + if (!hasCriticalMemoryPressure()) return; + memoryPressureExceeded = true; + abortController.abort(); + }, 1000); + interval.unref?.(); + + try { + const result = await operation(); + if (memoryPressureExceeded) { + throw new Error(VIDEO_MEMORY_PRESSURE_ERROR); + } + return result; + } catch (error) { + if (memoryPressureExceeded) { + throw new Error(VIDEO_MEMORY_PRESSURE_ERROR); + } + throw error; + } finally { + clearInterval(interval); + } +} + video.get("/status", (c) => { const jobs = getAllJobs(); const resources = getSystemResources(); @@ -1618,15 +1655,30 @@ async function streamConcatFiles( inputPaths: string[], outputPath: string, ): Promise { - const { open, readFile } = await import("node:fs/promises"); - const handle = await open(outputPath, "w"); + const writer = file(outputPath).writer(); + let lastMemoryCheckAt = 0; try { for (const filePath of inputPaths) { - const data = await readFile(filePath); - await handle.write(data); + const reader = file(filePath).stream().getReader(); + try { + while (true) { + const { done, value } = await reader.read(); + if (done) break; + writer.write(value); + const now = Date.now(); + if (now - lastMemoryCheckAt >= 1000) { + lastMemoryCheckAt = now; + if (hasCriticalMemoryPressure()) { + throw new Error(VIDEO_MEMORY_PRESSURE_ERROR); + } + } + } + } finally { + reader.releaseLock(); + } } } finally { - await handle.close(); + await writer.end(); } } @@ -1639,16 +1691,111 @@ function redactPresignedUrl(url: string): string { } } -async function downloadUrlToFile(url: string, destPath: string): Promise { - const resp = await fetch(url, { signal: AbortSignal.timeout(120_000) }); +class MediaDownloadError extends Error { + constructor( + message: string, + readonly retryable: boolean, + ) { + super(message); + } +} + +function isRetryableDownloadStatus(status: number): boolean { + return ( + status === 408 || + status === 425 || + status === 429 || + status === 500 || + status === 502 || + status === 503 || + status === 504 + ); +} + +async function downloadUrlToFileOnce( + url: string, + destPath: string, +): Promise { + const abortController = new AbortController(); + const resp = await fetch(url, { + signal: AbortSignal.any([ + abortController.signal, + AbortSignal.timeout(120_000), + ]), + }); if (!resp.ok) { - throw new Error( + await resp.body?.cancel().catch(() => {}); + throw new MediaDownloadError( `Download failed (${resp.status}): ${redactPresignedUrl(url)}`, + isRetryableDownloadStatus(resp.status), + ); + } + if (!resp.body) { + throw new MediaDownloadError( + `Download returned no body: ${redactPresignedUrl(url)}`, + true, ); } - const data = Buffer.from(await resp.arrayBuffer()); - const { writeFile } = await import("node:fs/promises"); - await writeFile(destPath, data); + + const reader = resp.body.getReader(); + const writer = file(destPath).writer(); + let lastMemoryCheckAt = 0; + let failure: unknown; + let completed = false; + try { + while (true) { + const { done, value } = await reader.read(); + if (done) break; + writer.write(value); + const now = Date.now(); + if (now - lastMemoryCheckAt >= 1000) { + lastMemoryCheckAt = now; + if (hasCriticalMemoryPressure()) { + throw new Error(VIDEO_MEMORY_PRESSURE_ERROR); + } + } + } + completed = true; + } catch (error) { + failure = error; + abortController.abort(); + await reader.cancel().catch(() => {}); + } finally { + reader.releaseLock(); + await writer.end(); + } + + if (!completed) { + const { rm } = await import("node:fs/promises"); + await rm(destPath, { force: true }).catch(() => {}); + throw failure ?? new Error("Download failed while streaming response body"); + } +} + +async function downloadUrlToFile(url: string, destPath: string): Promise { + let lastError: Error | undefined; + + for (let attempt = 0; attempt < SEGMENT_DOWNLOAD_MAX_ATTEMPTS; attempt++) { + try { + await downloadUrlToFileOnce(url, destPath); + return; + } catch (error) { + if (isBusyError(error)) throw error; + + const downloadError = + error instanceof Error ? error : new Error(String(error)); + if (error instanceof MediaDownloadError && !error.retryable) { + throw downloadError; + } + lastError = downloadError; + } + + if (attempt < SEGMENT_DOWNLOAD_MAX_ATTEMPTS - 1) { + await Bun.sleep(SEGMENT_DOWNLOAD_RETRY_BASE_MS * 2 ** attempt); + } + } + + throw lastError ?? new Error("Download failed"); } async function downloadSegmentsBatchTracked( @@ -1657,30 +1804,36 @@ async function downloadSegmentsBatchTracked( jobId: string, progressBase: number, progressRange: number, -): Promise { +): Promise { const { join } = await import("node:path"); let completed = 0; - let failed = 0; + let fatalError: Error | undefined; const total = urls.length; + const indexWidth = Math.max(3, String(total).length); + const outputPaths = new Array(total); const pending = [...urls.entries()]; const CONCURRENCY = 10; async function worker() { - while (pending.length > 0) { + while (pending.length > 0 && !fatalError) { const entry = pending.shift(); if (!entry) break; const [i, url] = entry; try { - await downloadUrlToFile( - url, - join(dir, `segment_${String(i + 1).padStart(3, "0")}.m4s`), + const outputPath = join( + dir, + `segment_${String(i + 1).padStart(indexWidth, "0")}.m4s`, ); + outputPaths[i] = outputPath; + await downloadUrlToFile(url, outputPath); } catch (err) { - failed++; + fatalError = err instanceof Error ? err : new Error(String(err)); + pending.length = 0; console.error( `[mux-segments] Failed to download segment ${i + 1}/${total}:`, err instanceof Error ? err.message : err, ); + break; } completed++; if (total > 0) { @@ -1696,7 +1849,8 @@ async function downloadSegmentsBatchTracked( await Promise.all( Array.from({ length: Math.min(CONCURRENCY, total) }, () => worker()), ); - return failed; + if (fatalError) throw fatalError; + return outputPaths; } function sendCurrentJobWebhook(jobId: string): void { @@ -1720,7 +1874,7 @@ async function muxSegmentsAsync( audioSegmentUrls: string[] | null, ): Promise { const { ensureTempDir } = await import("../lib/temp-files"); - const { mkdir, readdir } = await import("node:fs/promises"); + const { mkdir, rm } = await import("node:fs/promises"); const { join } = await import("node:path"); const workDir = join( @@ -1731,8 +1885,16 @@ async function muxSegmentsAsync( const abortController = new AbortController(); updateJob(jobId, { abortController }); let outputUploadStarted = false; + const startedAt = Date.now(); try { + logVideoEvent("video_mux_started", { + jobId, + videoId, + videoSegmentCount: videoSegmentUrls.length, + audioSegmentCount: audioSegmentUrls?.length ?? 0, + ...getVideoCapacitySnapshot(), + }); await ensureTempDir(); updateJob(jobId, { phase: "downloading", progress: 0 }); sendCurrentJobWebhook(jobId); @@ -1747,7 +1909,7 @@ async function muxSegmentsAsync( updateJob(jobId, { phase: "downloading", progress: 5 }); sendCurrentJobWebhook(jobId); - const videoFailed = await downloadSegmentsBatchTracked( + const videoSegmentFiles = await downloadSegmentsBatchTracked( videoSegmentUrls, videoDir, jobId, @@ -1755,46 +1917,22 @@ async function muxSegmentsAsync( 45, ); - if (videoFailed > 0) { - const failRatio = videoFailed / videoSegmentUrls.length; - if (failRatio >= 0.5) { - throw new Error( - `Too many video segments failed: ${videoFailed}/${videoSegmentUrls.length} (${Math.round(failRatio * 100)}%)`, - ); - } - console.warn( - `[mux-segments] ${videoFailed}/${videoSegmentUrls.length} video segments failed to download for ${videoId}`, - ); - } - - let audioInput = + const audioInput = audioInitUrl !== null && audioSegmentUrls !== null && audioSegmentUrls.length > 0 ? { initUrl: audioInitUrl, segmentUrls: audioSegmentUrls } : null; + let audioSegmentFiles: string[] = []; if (audioInput) { await downloadUrlToFile(audioInput.initUrl, join(audioDir, "init.mp4")); - const audioFailed = await downloadSegmentsBatchTracked( + audioSegmentFiles = await downloadSegmentsBatchTracked( audioInput.segmentUrls, audioDir, jobId, 50, 10, ); - if (audioFailed > 0) { - const audioFailRatio = audioFailed / audioInput.segmentUrls.length; - if (audioFailRatio >= 0.5) { - console.warn( - `[mux-segments] ${audioFailed}/${audioInput.segmentUrls.length} audio segments failed for ${videoId} (${Math.round(audioFailRatio * 100)}%), proceeding without audio`, - ); - audioInput = null; - } else { - console.warn( - `[mux-segments] ${audioFailed}/${audioInput.segmentUrls.length} audio segments failed to download for ${videoId}`, - ); - } - } } updateJob(jobId, { phase: "processing", progress: 60 }); @@ -1802,55 +1940,57 @@ async function muxSegmentsAsync( const combinedVideoPath = join(workDir, "combined_video.mp4"); const videoInitPath = join(videoDir, "init.mp4"); - const videoSegmentFiles = (await readdir(videoDir)) - .filter((f) => f.endsWith(".m4s")) - .sort() - .map((f) => join(videoDir, f)); await streamConcatFiles( [videoInitPath, ...videoSegmentFiles], combinedVideoPath, ); + await rm(videoDir, { recursive: true, force: true }); - const videoOnlyPath = join(workDir, "video_only.mp4"); - await muxMediaTracksToMp4( - combinedVideoPath, - null, - videoOnlyPath, - abortController.signal, - ); - - let resultPath: string; + let combinedAudioPath: string | null = null; if (audioInput) { - const combinedAudioPath = join(workDir, "combined_audio.mp4"); + combinedAudioPath = join(workDir, "combined_audio.mp4"); const audioInitPath = join(audioDir, "init.mp4"); - const audioSegmentFiles = (await readdir(audioDir)) - .filter((f) => f.endsWith(".m4s")) - .sort() - .map((f) => join(audioDir, f)); await streamConcatFiles( [audioInitPath, ...audioSegmentFiles], combinedAudioPath, ); + await rm(audioDir, { recursive: true, force: true }); + } - resultPath = join(workDir, "result.mp4"); - await muxMediaTracksToMp4( - videoOnlyPath, + if (hasCriticalMemoryPressure()) { + throw new Error(VIDEO_MEMORY_PRESSURE_ERROR); + } + logVideoEvent("video_mux_inputs_materialized", { + jobId, + videoId, + combinedVideoSize: file(combinedVideoPath).size, + combinedAudioSize: combinedAudioPath ? file(combinedAudioPath).size : 0, + resources: getSystemResources(), + }); + + const resultPath = join(workDir, "result.mp4"); + await withMuxMemoryGuard(abortController, () => + muxMediaTracksToMp4( + combinedVideoPath, combinedAudioPath, resultPath, abortController.signal, - ); - } else { - resultPath = join(workDir, "result.mp4"); - await muxMediaTracksToMp4( - videoOnlyPath, - null, - resultPath, - abortController.signal, - ); + ), + ); + await rm(combinedVideoPath, { force: true }); + if (combinedAudioPath) { + await rm(combinedAudioPath, { force: true }); } + logVideoEvent("video_mux_output_ready", { + jobId, + videoId, + outputSize: file(resultPath).size, + durationMs: Date.now() - startedAt, + resources: getSystemResources(), + }); updateJob(jobId, { phase: "uploading", progress: 80 }); sendCurrentJobWebhook(jobId); @@ -1900,9 +2040,23 @@ async function muxSegmentsAsync( metadata, }); sendCurrentJobWebhook(jobId); + logVideoEvent("video_mux_succeeded", { + jobId, + videoId, + durationMs: Date.now() - startedAt, + metadata, + resources: getSystemResources(), + }); setTimeout(() => deleteJob(jobId), 5 * 60 * 1000); } catch (error: unknown) { + logVideoEvent("video_mux_failed", { + jobId, + videoId, + durationMs: Date.now() - startedAt, + error: error instanceof Error ? error.message : String(error), + resources: getSystemResources(), + }); if (!outputUploadStarted) { await abortStorageUploadTarget(outputUpload).catch((abortError) => { console.warn( @@ -1918,7 +2072,6 @@ async function muxSegmentsAsync( }); sendCurrentJobWebhook(jobId); } finally { - const { rm } = await import("node:fs/promises"); await rm(workDir, { recursive: true, force: true }).catch(() => {}); } } From d7eaffee864ec8c230d1bf15f12c1bdec7fbff9c Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Wed, 15 Jul 2026 17:51:17 +0300 Subject: [PATCH 04/17] chore: pin media server runtime --- apps/media-server/Dockerfile | 2 +- apps/media-server/Dockerfile.standalone | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/apps/media-server/Dockerfile b/apps/media-server/Dockerfile index 754bd9f2b50..5b754ac7bc8 100644 --- a/apps/media-server/Dockerfile +++ b/apps/media-server/Dockerfile @@ -1,4 +1,4 @@ -FROM oven/bun:1 +FROM oven/bun:1.3.14 RUN apt-get update \ && apt-get install -y --no-install-recommends ffmpeg \ diff --git a/apps/media-server/Dockerfile.standalone b/apps/media-server/Dockerfile.standalone index 105f62f3338..281e406ef79 100644 --- a/apps/media-server/Dockerfile.standalone +++ b/apps/media-server/Dockerfile.standalone @@ -1,4 +1,4 @@ -FROM oven/bun:1 +FROM oven/bun:1.3.14 RUN apt-get update \ && apt-get install -y --no-install-recommends ffmpeg \ From 12e3829dcbcc56833eee1811fd311635f0ef126c Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Wed, 15 Jul 2026 21:23:36 +0300 Subject: [PATCH 05/17] fix: use media server image healthcheck --- docker-compose.coolify.yml | 6 ------ docker-compose.template.yml | 6 ------ docker-compose.yml | 6 ------ 3 files changed, 18 deletions(-) diff --git a/docker-compose.coolify.yml b/docker-compose.coolify.yml index 8c89312e36d..077ef34962f 100644 --- a/docker-compose.coolify.yml +++ b/docker-compose.coolify.yml @@ -46,12 +46,6 @@ services: environment: PORT: 3456 MEDIA_SERVER_WEBHOOK_SECRET: ${MEDIA_SERVER_WEBHOOK_SECRET} - healthcheck: - test: ["CMD", "wget", "-qO-", "http://localhost:3456/health"] - interval: 30s - timeout: 10s - retries: 3 - start_period: 10s networks: - cap-network diff --git a/docker-compose.template.yml b/docker-compose.template.yml index e21f40d2802..4351f4df876 100644 --- a/docker-compose.template.yml +++ b/docker-compose.template.yml @@ -60,12 +60,6 @@ services: MEDIA_SERVER_WEBHOOK_SECRET: fe337b52749070bb7b5d2c78cff9948439ea73cbc1869ba39d350e6c24db53b1 ports: - "3456:3456" - healthcheck: - test: ["CMD", "wget", "-qO-", "http://localhost:3456/health"] - interval: 30s - timeout: 10s - retries: 3 - start_period: 10s # Local S3 Strorage minio: diff --git a/docker-compose.yml b/docker-compose.yml index 50add5ca7f6..c74c50a2d5b 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -48,12 +48,6 @@ services: environment: PORT: 3456 MEDIA_SERVER_WEBHOOK_SECRET: ${MEDIA_SERVER_WEBHOOK_SECRET:-fedcba9876543210fedcba9876543210fedcba9876543210fedcba9876543210} - healthcheck: - test: ["CMD", "wget", "-qO-", "http://localhost:3456/health"] - interval: 30s - timeout: 10s - retries: 3 - start_period: 10s networks: - cap-network From 06f610d27d27e08ca4f187335d489132484ac217 Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Wed, 15 Jul 2026 21:27:15 +0300 Subject: [PATCH 06/17] fix: preserve media health response --- .../src/__tests__/routes/health.test.ts | 4 ++++ apps/media-server/src/routes/health.ts | 17 ++++++++++++----- 2 files changed, 16 insertions(+), 5 deletions(-) diff --git a/apps/media-server/src/__tests__/routes/health.test.ts b/apps/media-server/src/__tests__/routes/health.test.ts index 5d6d04e808d..87d15deb4d5 100644 --- a/apps/media-server/src/__tests__/routes/health.test.ts +++ b/apps/media-server/src/__tests__/routes/health.test.ts @@ -16,5 +16,9 @@ describe("GET /health", () => { expect(typeof data.system.containerMemoryLimitMB).toBe("number"); expect(typeof data.system.memoryPressure).toBe("number"); expect(typeof data.system.processRssMB).toBe("number"); + expect(typeof data.system.totalMemoryMB).toBe("number"); + expect(typeof data.system.freeMemoryMB).toBe("number"); + expect(typeof data.system.memoryUsagePercent).toBe("number"); + expect(typeof data.system.uptimeSeconds).toBe("number"); }); }); diff --git a/apps/media-server/src/routes/health.ts b/apps/media-server/src/routes/health.ts index f8a640af688..5386ef060f3 100644 --- a/apps/media-server/src/routes/health.ts +++ b/apps/media-server/src/routes/health.ts @@ -7,7 +7,12 @@ const health = new Hono(); health.get("/", (c) => { const mediaEngine = getMediaEngineStatus(); - const loadAvg = os.loadavg(); + const [, loadAvg5m, loadAvg15m] = os.loadavg(); + const totalMemoryMB = Math.round(os.totalmem() / (1024 * 1024)); + const freeMemoryMB = Math.round(os.freemem() / (1024 * 1024)); + const memoryUsagePercent = Math.round( + (1 - os.freemem() / os.totalmem()) * 100, + ); const resources = getSystemResources(); return c.json({ @@ -20,10 +25,12 @@ health.get("/", (c) => { }, system: { ...resources, - loadAvg1m: loadAvg[0], - loadAvg5m: loadAvg[1], - loadAvg15m: loadAvg[2], - uptimeSeconds: Math.round(process.uptime()), + loadAvg5m, + loadAvg15m, + totalMemoryMB, + freeMemoryMB, + memoryUsagePercent, + uptimeSeconds: Math.round(os.uptime()), }, }); }); From 3f395c150b0e023a28c764a43a674e83183e8480 Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Wed, 15 Jul 2026 21:27:15 +0300 Subject: [PATCH 07/17] fix: reuse detected CPU capacity --- apps/media-server/src/lib/job-manager.ts | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/apps/media-server/src/lib/job-manager.ts b/apps/media-server/src/lib/job-manager.ts index 8c6c513c864..c998af9f912 100644 --- a/apps/media-server/src/lib/job-manager.ts +++ b/apps/media-server/src/lib/job-manager.ts @@ -96,11 +96,10 @@ function getCpuCapacity(): number { } function getCpuPressure(cpuCapacity: number, loadAvg1m: number): number { - const cpuLimit = getContainerCpuLimit(); const usageMicros = getContainerCpuUsageMicros(); const now = performance.now(); - if (usageMicros > 0 && cpuLimit > 0) { + if (usageMicros > 0) { if ( previousContainerCpuUsageMicros > 0 && now - previousContainerCpuSampleAt >= CPU_SAMPLE_MIN_INTERVAL_MS From b6fe89466d01833b6818360626f9602d91748a26 Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Wed, 15 Jul 2026 21:27:15 +0300 Subject: [PATCH 08/17] fix: cancel failed segment batches --- ...edia-routes-real-world.integration.test.ts | 19 +++++++++- apps/media-server/src/routes/video.ts | 37 ++++++++++++------- 2 files changed, 42 insertions(+), 14 deletions(-) diff --git a/apps/media-server/src/__tests__/lib/media-routes-real-world.integration.test.ts b/apps/media-server/src/__tests__/lib/media-routes-real-world.integration.test.ts index 73d10ce2c67..9fa8591f693 100644 --- a/apps/media-server/src/__tests__/lib/media-routes-real-world.integration.test.ts +++ b/apps/media-server/src/__tests__/lib/media-routes-real-world.integration.test.ts @@ -35,6 +35,7 @@ let tempDir = ""; const uploadedArtifacts = new Map(); let transientFixtureFailures = 0; let permanentFixtureFailures = 0; +let slowFixtureCancellations = 0; function fileUrl(path: string) { return pathToFileURL(path).toString(); @@ -144,6 +145,18 @@ beforeAll(async () => { permanentFixtureFailures++; return new Response("Unavailable", { status: 503 }); } + if (url.pathname === "/fixtures/slow-segment.m4s") { + return new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(new Uint8Array(1024)); + }, + cancel() { + slowFixtureCancellations++; + }, + }), + ); + } if ( request.method === "GET" && url.pathname === "/fixtures/transient-no-audio.mp4" && @@ -374,7 +387,10 @@ describe("media routes real-world integration tests", () => { userId: "failed-segment-mux-user", outputPresignedUrl: uploadUrl("failed-segment-output.mp4"), videoInitUrl: fixtureUrl("test-no-audio.mp4"), - videoSegmentUrls: [`${baseUrl}/fixtures/permanent-unavailable.m4s`], + videoSegmentUrls: [ + `${baseUrl}/fixtures/permanent-unavailable.m4s`, + `${baseUrl}/fixtures/slow-segment.m4s`, + ], }), ); @@ -385,6 +401,7 @@ describe("media routes real-world integration tests", () => { expect(job.phase).toBe("error"); expect(job.error).toContain("503"); expect(permanentFixtureFailures).toBe(3); + expect(slowFixtureCancellations).toBe(1); expect(uploadedArtifacts.has("/uploads/failed-segment-output.mp4")).toBe( false, ); diff --git a/apps/media-server/src/routes/video.ts b/apps/media-server/src/routes/video.ts index 4b4040c1cad..3d444d44722 100644 --- a/apps/media-server/src/routes/video.ts +++ b/apps/media-server/src/routes/video.ts @@ -1715,13 +1715,14 @@ function isRetryableDownloadStatus(status: number): boolean { async function downloadUrlToFileOnce( url: string, destPath: string, + abortSignal?: AbortSignal, ): Promise { const abortController = new AbortController(); + const timeoutSignal = AbortSignal.timeout(120_000); const resp = await fetch(url, { - signal: AbortSignal.any([ - abortController.signal, - AbortSignal.timeout(120_000), - ]), + signal: abortSignal + ? AbortSignal.any([abortController.signal, abortSignal, timeoutSignal]) + : AbortSignal.any([abortController.signal, timeoutSignal]), }); if (!resp.ok) { await resp.body?.cancel().catch(() => {}); @@ -1772,14 +1773,20 @@ async function downloadUrlToFileOnce( } } -async function downloadUrlToFile(url: string, destPath: string): Promise { +async function downloadUrlToFile( + url: string, + destPath: string, + abortSignal?: AbortSignal, +): Promise { let lastError: Error | undefined; for (let attempt = 0; attempt < SEGMENT_DOWNLOAD_MAX_ATTEMPTS; attempt++) { + abortSignal?.throwIfAborted(); try { - await downloadUrlToFileOnce(url, destPath); + await downloadUrlToFileOnce(url, destPath, abortSignal); return; } catch (error) { + if (abortSignal?.aborted) throw error; if (isBusyError(error)) throw error; const downloadError = @@ -1812,6 +1819,7 @@ async function downloadSegmentsBatchTracked( const indexWidth = Math.max(3, String(total).length); const outputPaths = new Array(total); const pending = [...urls.entries()]; + const batchAbortController = new AbortController(); const CONCURRENCY = 10; async function worker() { @@ -1825,14 +1833,17 @@ async function downloadSegmentsBatchTracked( `segment_${String(i + 1).padStart(indexWidth, "0")}.m4s`, ); outputPaths[i] = outputPath; - await downloadUrlToFile(url, outputPath); + await downloadUrlToFile(url, outputPath, batchAbortController.signal); } catch (err) { - fatalError = err instanceof Error ? err : new Error(String(err)); - pending.length = 0; - console.error( - `[mux-segments] Failed to download segment ${i + 1}/${total}:`, - err instanceof Error ? err.message : err, - ); + if (!fatalError) { + fatalError = err instanceof Error ? err : new Error(String(err)); + pending.length = 0; + batchAbortController.abort(); + console.error( + `[mux-segments] Failed to download segment ${i + 1}/${total}:`, + err instanceof Error ? err.message : err, + ); + } break; } completed++; From 0ee14440dc3f30e229a27d2f4b59d24aa4c820da Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Wed, 15 Jul 2026 21:30:02 +0300 Subject: [PATCH 09/17] fix: harden segment download cleanup --- apps/media-server/src/routes/video.ts | 19 +++++++++++-------- 1 file changed, 11 insertions(+), 8 deletions(-) diff --git a/apps/media-server/src/routes/video.ts b/apps/media-server/src/routes/video.ts index 3d444d44722..baef9385cd8 100644 --- a/apps/media-server/src/routes/video.ts +++ b/apps/media-server/src/routes/video.ts @@ -1742,7 +1742,6 @@ async function downloadUrlToFileOnce( const writer = file(destPath).writer(); let lastMemoryCheckAt = 0; let failure: unknown; - let completed = false; try { while (true) { const { done, value } = await reader.read(); @@ -1756,20 +1755,23 @@ async function downloadUrlToFileOnce( } } } - completed = true; } catch (error) { failure = error; abortController.abort(); await reader.cancel().catch(() => {}); } finally { reader.releaseLock(); - await writer.end(); + try { + await writer.end(); + } catch (error) { + failure ??= error; + } } - if (!completed) { + if (failure !== undefined) { const { rm } = await import("node:fs/promises"); await rm(destPath, { force: true }).catch(() => {}); - throw failure ?? new Error("Download failed while streaming response body"); + throw failure; } } @@ -1819,12 +1821,13 @@ async function downloadSegmentsBatchTracked( const indexWidth = Math.max(3, String(total).length); const outputPaths = new Array(total); const pending = [...urls.entries()]; + let pendingIndex = 0; const batchAbortController = new AbortController(); const CONCURRENCY = 10; async function worker() { - while (pending.length > 0 && !fatalError) { - const entry = pending.shift(); + while (pendingIndex < pending.length && !fatalError) { + const entry = pending[pendingIndex++]; if (!entry) break; const [i, url] = entry; try { @@ -1837,7 +1840,7 @@ async function downloadSegmentsBatchTracked( } catch (err) { if (!fatalError) { fatalError = err instanceof Error ? err : new Error(String(err)); - pending.length = 0; + pendingIndex = pending.length; batchAbortController.abort(); console.error( `[mux-segments] Failed to download segment ${i + 1}/${total}:`, From e8315eae87029c24298c9a728be1a4168da9a516 Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Wed, 15 Jul 2026 21:34:24 +0300 Subject: [PATCH 10/17] fix: snapshot health memory metrics --- apps/media-server/src/routes/health.ts | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/apps/media-server/src/routes/health.ts b/apps/media-server/src/routes/health.ts index 5386ef060f3..0a2209ec9cd 100644 --- a/apps/media-server/src/routes/health.ts +++ b/apps/media-server/src/routes/health.ts @@ -8,10 +8,12 @@ const health = new Hono(); health.get("/", (c) => { const mediaEngine = getMediaEngineStatus(); const [, loadAvg5m, loadAvg15m] = os.loadavg(); - const totalMemoryMB = Math.round(os.totalmem() / (1024 * 1024)); - const freeMemoryMB = Math.round(os.freemem() / (1024 * 1024)); + const totalMemoryBytes = os.totalmem(); + const freeMemoryBytes = os.freemem(); + const totalMemoryMB = Math.round(totalMemoryBytes / (1024 * 1024)); + const freeMemoryMB = Math.round(freeMemoryBytes / (1024 * 1024)); const memoryUsagePercent = Math.round( - (1 - os.freemem() / os.totalmem()) * 100, + (1 - freeMemoryBytes / totalMemoryBytes) * 100, ); const resources = getSystemResources(); From 6d6393fa6a05ee7bae274f763fc830a803675bb6 Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Wed, 15 Jul 2026 21:34:24 +0300 Subject: [PATCH 11/17] fix: normalize segment download errors --- apps/media-server/src/routes/video.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/apps/media-server/src/routes/video.ts b/apps/media-server/src/routes/video.ts index baef9385cd8..bbe021c144f 100644 --- a/apps/media-server/src/routes/video.ts +++ b/apps/media-server/src/routes/video.ts @@ -1771,7 +1771,9 @@ async function downloadUrlToFileOnce( if (failure !== undefined) { const { rm } = await import("node:fs/promises"); await rm(destPath, { force: true }).catch(() => {}); - throw failure; + throw failure instanceof Error + ? failure + : new Error("Download failed while streaming response body"); } } From f0ee4b4c7dda9383c2ea3540eadcab0bb4c2969e Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Wed, 15 Jul 2026 21:36:49 +0300 Subject: [PATCH 12/17] fix: use Bun for compose healthchecks --- docker-compose.coolify.yml | 6 ++++++ docker-compose.template.yml | 6 ++++++ docker-compose.yml | 6 ++++++ 3 files changed, 18 insertions(+) diff --git a/docker-compose.coolify.yml b/docker-compose.coolify.yml index 077ef34962f..85c2ef7fe5b 100644 --- a/docker-compose.coolify.yml +++ b/docker-compose.coolify.yml @@ -46,6 +46,12 @@ services: environment: PORT: 3456 MEDIA_SERVER_WEBHOOK_SECRET: ${MEDIA_SERVER_WEBHOOK_SECRET} + healthcheck: + test: ["CMD", "bun", "-e", "const response = await fetch('http://localhost:3456/health'); process.exit(response.ok ? 0 : 1)"] + interval: 30s + timeout: 10s + retries: 3 + start_period: 10s networks: - cap-network diff --git a/docker-compose.template.yml b/docker-compose.template.yml index 4351f4df876..dfac3c4fdef 100644 --- a/docker-compose.template.yml +++ b/docker-compose.template.yml @@ -60,6 +60,12 @@ services: MEDIA_SERVER_WEBHOOK_SECRET: fe337b52749070bb7b5d2c78cff9948439ea73cbc1869ba39d350e6c24db53b1 ports: - "3456:3456" + healthcheck: + test: ["CMD", "bun", "-e", "const response = await fetch('http://localhost:3456/health'); process.exit(response.ok ? 0 : 1)"] + interval: 30s + timeout: 10s + retries: 3 + start_period: 10s # Local S3 Strorage minio: diff --git a/docker-compose.yml b/docker-compose.yml index c74c50a2d5b..9c1f8f19c11 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -48,6 +48,12 @@ services: environment: PORT: 3456 MEDIA_SERVER_WEBHOOK_SECRET: ${MEDIA_SERVER_WEBHOOK_SECRET:-fedcba9876543210fedcba9876543210fedcba9876543210fedcba9876543210} + healthcheck: + test: ["CMD", "bun", "-e", "const response = await fetch('http://localhost:3456/health'); process.exit(response.ok ? 0 : 1)"] + interval: 30s + timeout: 10s + retries: 3 + start_period: 10s networks: - cap-network From 145964ba599f21fbb88eebcadf123216092c594f Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Wed, 15 Jul 2026 21:40:47 +0300 Subject: [PATCH 13/17] fix: make media healthchecks deterministic --- apps/media-server/Dockerfile | 2 +- apps/media-server/Dockerfile.standalone | 2 +- docker-compose.coolify.yml | 2 +- docker-compose.template.yml | 2 +- docker-compose.yml | 2 +- 5 files changed, 5 insertions(+), 5 deletions(-) diff --git a/apps/media-server/Dockerfile b/apps/media-server/Dockerfile index 5b754ac7bc8..69c27c88dd3 100644 --- a/apps/media-server/Dockerfile +++ b/apps/media-server/Dockerfile @@ -15,6 +15,6 @@ ENV PORT=3456 EXPOSE 3456 HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \ - CMD ["bun", "-e", "const response = await fetch(`http://localhost:${process.env.PORT || 3456}/health`); process.exit(response.ok ? 0 : 1)"] + 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", "run", "src/index.ts"] diff --git a/apps/media-server/Dockerfile.standalone b/apps/media-server/Dockerfile.standalone index 281e406ef79..201267eff8d 100644 --- a/apps/media-server/Dockerfile.standalone +++ b/apps/media-server/Dockerfile.standalone @@ -15,6 +15,6 @@ ENV PORT=3456 EXPOSE 3456 HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \ - CMD ["bun", "-e", "const response = await fetch(`http://localhost:${process.env.PORT || 3456}/health`); process.exit(response.ok ? 0 : 1)"] + 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", "run", "src/index.ts"] diff --git a/docker-compose.coolify.yml b/docker-compose.coolify.yml index 85c2ef7fe5b..d14469468fb 100644 --- a/docker-compose.coolify.yml +++ b/docker-compose.coolify.yml @@ -47,7 +47,7 @@ services: PORT: 3456 MEDIA_SERVER_WEBHOOK_SECRET: ${MEDIA_SERVER_WEBHOOK_SECRET} healthcheck: - test: ["CMD", "bun", "-e", "const response = await fetch('http://localhost:3456/health'); process.exit(response.ok ? 0 : 1)"] + test: ["CMD", "bun", "-e", "fetch('http://127.0.0.1:3456/health').then((response) => process.exit(response.ok ? 0 : 1)).catch(() => process.exit(1))"] interval: 30s timeout: 10s retries: 3 diff --git a/docker-compose.template.yml b/docker-compose.template.yml index dfac3c4fdef..1360b3d8715 100644 --- a/docker-compose.template.yml +++ b/docker-compose.template.yml @@ -61,7 +61,7 @@ services: ports: - "3456:3456" healthcheck: - test: ["CMD", "bun", "-e", "const response = await fetch('http://localhost:3456/health'); process.exit(response.ok ? 0 : 1)"] + test: ["CMD", "bun", "-e", "fetch('http://127.0.0.1:3456/health').then((response) => process.exit(response.ok ? 0 : 1)).catch(() => process.exit(1))"] interval: 30s timeout: 10s retries: 3 diff --git a/docker-compose.yml b/docker-compose.yml index 9c1f8f19c11..a7f62ad011a 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -49,7 +49,7 @@ services: PORT: 3456 MEDIA_SERVER_WEBHOOK_SECRET: ${MEDIA_SERVER_WEBHOOK_SECRET:-fedcba9876543210fedcba9876543210fedcba9876543210fedcba9876543210} healthcheck: - test: ["CMD", "bun", "-e", "const response = await fetch('http://localhost:3456/health'); process.exit(response.ok ? 0 : 1)"] + test: ["CMD", "bun", "-e", "fetch('http://127.0.0.1:3456/health').then((response) => process.exit(response.ok ? 0 : 1)).catch(() => process.exit(1))"] interval: 30s timeout: 10s retries: 3 From 4606c96eb75ab1eeffa21910fb8b636737eea9f2 Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Wed, 15 Jul 2026 21:47:48 +0300 Subject: [PATCH 14/17] test: reset segment cancellation state --- .../__tests__/lib/media-routes-real-world.integration.test.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/apps/media-server/src/__tests__/lib/media-routes-real-world.integration.test.ts b/apps/media-server/src/__tests__/lib/media-routes-real-world.integration.test.ts index 9fa8591f693..38e0291cd4d 100644 --- a/apps/media-server/src/__tests__/lib/media-routes-real-world.integration.test.ts +++ b/apps/media-server/src/__tests__/lib/media-routes-real-world.integration.test.ts @@ -227,6 +227,7 @@ beforeEach(() => { uploadedArtifacts.clear(); transientFixtureFailures = 0; permanentFixtureFailures = 0; + slowFixtureCancellations = 0; }); afterAll(() => { From a74c3414c799da1033aab44730c7af0c3f1490df Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Wed, 15 Jul 2026 21:52:13 +0300 Subject: [PATCH 15/17] fix: reject unsafe manifest resource URLs --- .../lib/media-video.integration.test.ts | 60 +++++++++++++++++++ apps/media-server/src/lib/media-video.ts | 28 ++++++--- 2 files changed, 81 insertions(+), 7 deletions(-) diff --git a/apps/media-server/src/__tests__/lib/media-video.integration.test.ts b/apps/media-server/src/__tests__/lib/media-video.integration.test.ts index 0d565f37b62..f9b9237bc3b 100644 --- a/apps/media-server/src/__tests__/lib/media-video.integration.test.ts +++ b/apps/media-server/src/__tests__/lib/media-video.integration.test.ts @@ -340,6 +340,7 @@ describe("processVideo integration tests", () => { + media/ @@ -361,6 +362,9 @@ describe("processVideo integration tests", () => { expect(path.endsWith(".mpd")).toBe(true); expect(requests).toBe(2); + expect(content).toContain( + "https://cdn.example/video/media/?Policy=a&Signature=b", + ); expect(content).toContain("seg-1.m4s?Policy=a&Signature=b"); } finally { globalThis.fetch = originalFetch; @@ -392,6 +396,40 @@ describe("processVideo integration tests", () => { } }); + test("rejects local file references in remote DASH manifests", async () => { + const originalFetch = globalThis.fetch; + const manifestDir = mkdtempSync(join(tmpdir(), "cap-mpd-local-file-")); + + globalThis.fetch = (async () => + new Response( + ` + + + + file:///etc/ + + + + + + + `, + { status: 200, statusText: "OK" }, + )) as unknown as typeof fetch; + + try { + await expect( + materializeStreamingInput( + "https://cdn.example/video/manifest.mpd", + manifestDir, + ), + ).rejects.toThrow("Unsupported media resource protocol: file:"); + } finally { + globalThis.fetch = originalFetch; + rmSync(manifestDir, { recursive: true, force: true }); + } + }); + test("materializes signed HLS playlists with inherited query strings", async () => { const originalFetch = globalThis.fetch; const manifestDir = mkdtempSync(join(tmpdir(), "cap-hls-")); @@ -469,6 +507,28 @@ describe("processVideo integration tests", () => { } }); + test("rejects local file references in remote HLS playlists", async () => { + const originalFetch = globalThis.fetch; + const manifestDir = mkdtempSync(join(tmpdir(), "cap-hls-local-file-")); + + globalThis.fetch = (async () => + new Response( + ["#EXTM3U", "#EXTINF:1.0,", "file:///etc/passwd"].join("\n"), + )) as unknown as typeof fetch; + + try { + await expect( + materializeStreamingInput( + "https://cdn.example/video/manifest.m3u8", + manifestDir, + ), + ).rejects.toThrow("Unsupported media resource protocol: file:"); + } finally { + globalThis.fetch = originalFetch; + rmSync(manifestDir, { recursive: true, force: true }); + } + }); + test("waits for async cleanup before rejecting timed out work", async () => { let resolveCleanup: (() => void) | undefined; let settled = false; diff --git a/apps/media-server/src/lib/media-video.ts b/apps/media-server/src/lib/media-video.ts index 2552eccd8e9..86e7a2fb259 100644 --- a/apps/media-server/src/lib/media-video.ts +++ b/apps/media-server/src/lib/media-video.ts @@ -183,11 +183,14 @@ function resolveResourceUrl( baseUrl: string, query: string, ): string { - if (resource.startsWith("http://") || resource.startsWith("https://")) { - return withQuery(resource, query); + const resolved = new URL(resource, baseUrl); + if (resolved.protocol !== "http:" && resolved.protocol !== "https:") { + throw new Error( + `Unsupported media resource protocol: ${resolved.protocol}`, + ); } - return withQuery(new URL(resource, baseUrl).toString(), query); + return withQuery(resolved.toString(), query); } function getFetchSignal(timeoutMs: number, abortSignal?: AbortSignal) { @@ -318,9 +321,20 @@ export async function materializeMpdManifest( const baseUrl = new URL(".", parsedUrl).toString(); const query = parsedUrl.search; const filePath = join(dirPath, `${randomUUID()}.mpd`); + const rewrittenElements = content.replace( + /<(BaseURL|Location)(\b[^>]*)>([\s\S]*?)<\/\1>/gi, + (_, tag: string, attributes: string, resource: string) => { + const resolved = resolveResourceUrl( + decodeXmlAttribute(resource.trim()), + baseUrl, + query, + ); + return `<${tag}${attributes}>${escapeXmlAttribute(resolved)}`; + }, + ); - const rewritten = content.replace( - /(initialization|media)="([^"]+)"/g, + const rewritten = rewrittenElements.replace( + /(initialization|media|sourceURL|xlink:href|href)="([^"]+)"/gi, (_, attribute: string, resource: string) => { const resolved = resolveResourceUrl(resource, baseUrl, query); return `${attribute}="${escapeXmlAttribute(resolved)}"`; @@ -469,11 +483,11 @@ function getDashResourceBaseUrl( "BaseURL", ); const baseUrl = adaptationBaseUrl - ? new URL(adaptationBaseUrl, manifestBaseUrl).toString() + ? resolveResourceUrl(adaptationBaseUrl, manifestBaseUrl, "") : manifestBaseUrl; return representationBaseUrl - ? new URL(representationBaseUrl, baseUrl).toString() + ? resolveResourceUrl(representationBaseUrl, baseUrl, "") : baseUrl; } From d05fd7e07598a85b56f4124ea87e4d1f0ebb2c74 Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Wed, 15 Jul 2026 21:56:15 +0300 Subject: [PATCH 16/17] fix: scan materialized manifest protocols --- apps/media-server/src/lib/media-video.ts | 30 +++++++++++++++++------- 1 file changed, 22 insertions(+), 8 deletions(-) diff --git a/apps/media-server/src/lib/media-video.ts b/apps/media-server/src/lib/media-video.ts index 86e7a2fb259..4680fc1c340 100644 --- a/apps/media-server/src/lib/media-video.ts +++ b/apps/media-server/src/lib/media-video.ts @@ -1,5 +1,5 @@ import { randomUUID } from "node:crypto"; -import { mkdtemp, rm, writeFile } from "node:fs/promises"; +import { mkdtemp, readdir, rm, writeFile } from "node:fs/promises"; import { join } from "node:path"; import { type BunFile, file, spawn } from "bun"; import type { VideoMetadata } from "./job-manager"; @@ -847,25 +847,39 @@ export async function materializeStreamingInput( dirPath: string, abortSignal?: AbortSignal, ): Promise { + let inputPath: string; + if (isHlsUrl(videoUrl)) { - return await materializeHlsPlaylist( + inputPath = await materializeHlsPlaylist( videoUrl, dirPath, undefined, abortSignal, ); - } - - if (isMpdUrl(videoUrl)) { + } else if (isMpdUrl(videoUrl)) { try { - return await materializeMpdAsHlsPlaylist(videoUrl, dirPath, abortSignal); + inputPath = await materializeMpdAsHlsPlaylist( + videoUrl, + dirPath, + abortSignal, + ); } catch (err) { if (!shouldFallbackToGenericMpd(err)) throw err; - return await materializeMpdManifest(videoUrl, dirPath, abortSignal); + inputPath = await materializeMpdManifest(videoUrl, dirPath, abortSignal); + } + } else { + return videoUrl; + } + + for (const entry of await readdir(dirPath)) { + if (!entry.endsWith(".m3u8") && !entry.endsWith(".mpd")) continue; + const content = await file(join(dirPath, entry)).text(); + if (/\b(?:file|data):/i.test(content)) { + throw new Error("Unsupported manifest resource protocol"); } } - return videoUrl; + return inputPath; } async function drainStream( From 8ef77fc9e5fb67cb887c9fe7d46928a35e36047c Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Wed, 15 Jul 2026 22:00:16 +0300 Subject: [PATCH 17/17] fix: decode DASH resource attributes --- .../src/__tests__/lib/media-video.integration.test.ts | 4 +++- apps/media-server/src/lib/media-video.ts | 6 +++++- 2 files changed, 8 insertions(+), 2 deletions(-) diff --git a/apps/media-server/src/__tests__/lib/media-video.integration.test.ts b/apps/media-server/src/__tests__/lib/media-video.integration.test.ts index f9b9237bc3b..c5a3a82d58e 100644 --- a/apps/media-server/src/__tests__/lib/media-video.integration.test.ts +++ b/apps/media-server/src/__tests__/lib/media-video.integration.test.ts @@ -239,7 +239,7 @@ describe("processVideo integration tests", () => { globalThis.fetch = (async () => new Response( - '', + '', { status: 200, statusText: "OK" }, )) as unknown as typeof fetch; @@ -256,6 +256,8 @@ describe("processVideo integration tests", () => { expect(content).toContain( "chunk-$Number$.m4s?Policy=a&Signature=b&Key-Pair-Id=c", ); + expect(content).toContain("escaped.mp4?part=1&token=x"); + expect(content).not.toContain("&amp;"); } finally { globalThis.fetch = originalFetch; rmSync(manifestDir, { recursive: true, force: true }); diff --git a/apps/media-server/src/lib/media-video.ts b/apps/media-server/src/lib/media-video.ts index 4680fc1c340..a633a895800 100644 --- a/apps/media-server/src/lib/media-video.ts +++ b/apps/media-server/src/lib/media-video.ts @@ -336,7 +336,11 @@ export async function materializeMpdManifest( const rewritten = rewrittenElements.replace( /(initialization|media|sourceURL|xlink:href|href)="([^"]+)"/gi, (_, attribute: string, resource: string) => { - const resolved = resolveResourceUrl(resource, baseUrl, query); + const resolved = resolveResourceUrl( + decodeXmlAttribute(resource), + baseUrl, + query, + ); return `${attribute}="${escapeXmlAttribute(resolved)}"`; }, );