diff --git a/apps/media-server/Dockerfile b/apps/media-server/Dockerfile index 754bd9f2b50..69c27c88dd3 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 \ @@ -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 105f62f3338..201267eff8d 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 \ @@ -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/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__/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/__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..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 @@ -33,6 +33,9 @@ let baseUrl = ""; let tempDir = ""; const uploadedArtifacts = new Map(); +let transientFixtureFailures = 0; +let permanentFixtureFailures = 0; +let slowFixtureCancellations = 0; function fileUrl(path: string) { return pathToFileURL(path).toString(); @@ -138,12 +141,38 @@ 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 (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" && + 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 +198,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 +225,9 @@ beforeAll(async () => { beforeEach(() => { mock.restore(); uploadedArtifacts.clear(); + transientFixtureFailures = 0; + permanentFixtureFailures = 0; + slowFixtureCancellations = 0; }); afterAll(() => { @@ -320,6 +352,65 @@ 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`, + `${baseUrl}/fixtures/slow-segment.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(slowFixtureCancellations).toBe(1); + 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/__tests__/lib/media-video.integration.test.ts b/apps/media-server/src/__tests__/lib/media-video.integration.test.ts index 0d565f37b62..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 }); @@ -340,6 +342,7 @@ describe("processVideo integration tests", () => { + media/ @@ -361,6 +364,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 +398,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 +509,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/__tests__/routes/health.test.ts b/apps/media-server/src/__tests__/routes/health.test.ts index 68552268705..87d15deb4d5 100644 --- a/apps/media-server/src/__tests__/routes/health.test.ts +++ b/apps/media-server/src/__tests__/routes/health.test.ts @@ -12,5 +12,13 @@ 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"); + 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/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..c998af9f912 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,53 @@ 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 usageMicros = getContainerCpuUsageMicros(); + const now = performance.now(); - const bytes = Number.parseInt(rawValue, 10); + if (usageMicros > 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 +142,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 +157,7 @@ export function getMaxConcurrentVideoProcesses(): number { 1, Math.min( DEFAULT_MAX_CONCURRENT_VIDEO_PROCESSES, - Math.floor(cpuCount / 2), + Math.floor(getCpuCapacity() / 2), memoryBoundMax, ), ); @@ -156,11 +165,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 +181,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-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-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/lib/media-video.ts b/apps/media-server/src/lib/media-video.ts index 7b9fb8d9553..a633a895800 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"; @@ -183,11 +183,21 @@ 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) { + const timeoutSignal = AbortSignal.timeout(timeoutMs); + return abortSignal + ? AbortSignal.any([abortSignal, timeoutSignal]) + : timeoutSignal; } function redactUrl(value: string): string { @@ -227,12 +237,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 +269,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 +284,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 +304,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) { @@ -309,11 +321,26 @@ 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); + const resolved = resolveResourceUrl( + decodeXmlAttribute(resource), + baseUrl, + query, + ); return `${attribute}="${escapeXmlAttribute(resolved)}"`; }, ); @@ -460,11 +487,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; } @@ -656,9 +683,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,21 +849,41 @@ export async function materializeMpdAsHlsPlaylist( export async function materializeStreamingInput( videoUrl: string, dirPath: string, + abortSignal?: AbortSignal, ): Promise { - if (isHlsUrl(videoUrl)) { - return await materializeHlsPlaylist(videoUrl, dirPath); - } + let inputPath: string; - if (isMpdUrl(videoUrl)) { + if (isHlsUrl(videoUrl)) { + inputPath = await materializeHlsPlaylist( + videoUrl, + dirPath, + undefined, + abortSignal, + ); + } else if (isMpdUrl(videoUrl)) { try { - return await materializeMpdAsHlsPlaylist(videoUrl, dirPath); + inputPath = await materializeMpdAsHlsPlaylist( + videoUrl, + dirPath, + abortSignal, + ); } catch (err) { if (!shouldFallbackToGenericMpd(err)) throw err; - return await materializeMpdManifest(videoUrl, dirPath); + inputPath = await materializeMpdManifest(videoUrl, dirPath, abortSignal); } + } else { + return videoUrl; } - 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 inputPath; } async function drainStream( @@ -996,7 +1044,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 +1813,7 @@ async function uploadWithRetry( presignedUrl: string, contentType: string, contentLength: number, - bodyFactory: () => Blob | Uint8Array | ArrayBuffer | BunFile, + bodyFactory: () => Blob | BunFile, ): Promise { let lastError: Error | undefined; diff --git a/apps/media-server/src/routes/health.ts b/apps/media-server/src/routes/health.ts index 9501f32d5b5..0a2209ec9cd 100644 --- a/apps/media-server/src/routes/health.ts +++ b/apps/media-server/src/routes/health.ts @@ -1,32 +1,37 @@ 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 [, loadAvg5m, loadAvg15m] = os.loadavg(); + 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 - freeMemoryBytes / totalMemoryBytes) * 100, + ); + 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, - loadAvg1m: loadAvg[0], - loadAvg5m: loadAvg[1], - loadAvg15m: loadAvg[2], - totalMemoryMB: totalMemMB, - freeMemoryMB: freeMemMB, - memoryUsagePercent: Math.round(memoryUsagePercent * 100), + ...resources, + loadAvg5m, + loadAvg15m, + totalMemoryMB, + freeMemoryMB, + memoryUsagePercent, uptimeSeconds: Math.round(os.uptime()), }, }); diff --git a/apps/media-server/src/routes/video.ts b/apps/media-server/src/routes/video.ts index abc3158203e..bbe021c144f 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,122 @@ 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, + abortSignal?: AbortSignal, +): Promise { + const abortController = new AbortController(); + const timeoutSignal = AbortSignal.timeout(120_000); + const resp = await fetch(url, { + signal: abortSignal + ? AbortSignal.any([abortController.signal, abortSignal, timeoutSignal]) + : AbortSignal.any([abortController.signal, timeoutSignal]), + }); if (!resp.ok) { - throw new Error( + await resp.body?.cancel().catch(() => {}); + throw new MediaDownloadError( `Download failed (${resp.status}): ${redactPresignedUrl(url)}`, + isRetryableDownloadStatus(resp.status), ); } - const data = Buffer.from(await resp.arrayBuffer()); - const { writeFile } = await import("node:fs/promises"); - await writeFile(destPath, data); + if (!resp.body) { + throw new MediaDownloadError( + `Download returned no body: ${redactPresignedUrl(url)}`, + true, + ); + } + + const reader = resp.body.getReader(); + const writer = file(destPath).writer(); + let lastMemoryCheckAt = 0; + let failure: unknown; + 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); + } + } + } + } catch (error) { + failure = error; + abortController.abort(); + await reader.cancel().catch(() => {}); + } finally { + reader.releaseLock(); + try { + await writer.end(); + } catch (error) { + failure ??= error; + } + } + + if (failure !== undefined) { + const { rm } = await import("node:fs/promises"); + await rm(destPath, { force: true }).catch(() => {}); + throw failure instanceof Error + ? failure + : new Error("Download failed while streaming response body"); + } +} + +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, abortSignal); + return; + } catch (error) { + if (abortSignal?.aborted) throw 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 +1815,41 @@ 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()]; + let pendingIndex = 0; + const batchAbortController = new AbortController(); const CONCURRENCY = 10; async function worker() { - while (pending.length > 0) { - const entry = pending.shift(); + while (pendingIndex < pending.length && !fatalError) { + const entry = pending[pendingIndex++]; 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, batchAbortController.signal); } catch (err) { - failed++; - 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)); + pendingIndex = pending.length; + batchAbortController.abort(); + console.error( + `[mux-segments] Failed to download segment ${i + 1}/${total}:`, + err instanceof Error ? err.message : err, + ); + } + break; } completed++; if (total > 0) { @@ -1696,7 +1865,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 +1890,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 +1901,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 +1925,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 +1933,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 +1956,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 }); + } + + 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(), + }); - resultPath = join(workDir, "result.mp4"); - await muxMediaTracksToMp4( - videoOnlyPath, + 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 +2056,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 +2088,6 @@ async function muxSegmentsAsync( }); sendCurrentJobWebhook(jobId); } finally { - const { rm } = await import("node:fs/promises"); await rm(workDir, { recursive: true, force: true }).catch(() => {}); } } diff --git a/docker-compose.coolify.yml b/docker-compose.coolify.yml index 8c89312e36d..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", "wget", "-qO-", "http://localhost:3456/health"] + 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 e21f40d2802..1360b3d8715 100644 --- a/docker-compose.template.yml +++ b/docker-compose.template.yml @@ -61,7 +61,7 @@ services: ports: - "3456:3456" healthcheck: - test: ["CMD", "wget", "-qO-", "http://localhost:3456/health"] + 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 50add5ca7f6..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", "wget", "-qO-", "http://localhost:3456/health"] + 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