openvibers/openvibe.media Commits
Media: the live DVR snapshot is paced and big files leave the event loop (TargetDown 2026-10-13) (#102)
Changes
@@@ -103,18 +103,18 @@ function rebuildWebmFromMaster(masterPath, webmPath) {let ff;try { ff = spawn('ffmpeg', args, { stdio: 'ignore' }); } catch { return resolve(false); }const to = setTimeout(() => { try { ff.kill('SIGKILL'); } catch { /* */ } }, 45 * 60 * 1000);-- ff.on('close', (code) => {++ ff.on('close', async (code) => {clearTimeout(to);try {if (code === 0 && fs.existsSync(tmp) && fs.statSync(tmp).size > 1024) {-- fs.renameSync(tmp, webmPath);++ await fs.promises.rename(tmp, webmPath);return resolve(true);}} catch { /* fall through */ }-- try { fs.existsSync(tmp) && fs.unlinkSync(tmp); } catch { /* */ }++ await tools.removeFile(tmp);resolve(false);});-- ff.on('error', () => { clearTimeout(to); try { fs.existsSync(tmp) && fs.unlinkSync(tmp); } catch { /* */ } resolve(false); });++ ff.on('error', () => { clearTimeout(to); tools.removeFile(tmp).then(() => resolve(false)); });});}@@@ -203,9 +203,9 @@ async function _doFinalize(vodId, opts) {// Ephemeral clips-only recording: it existed only to serve live clips. Never// published — delete the file (+ any offloaded object) and its row.if (vod.clips_only) {-- try { if (filePath && fs.existsSync(filePath)) fs.unlinkSync(filePath); } catch { /* */ }-- try { tools.cleanupSeekableFile(filePath); } catch { /* */ }-- try { if (vod.master_file_path && fs.existsSync(vod.master_file_path)) fs.unlinkSync(vod.master_file_path); } catch { /* */ }++ await tools.removeFile(filePath);++ await tools.cleanupSeekableFile(filePath);++ await tools.removeFile(vod.master_file_path);try { if (vod.storage_provider && vod.storage_provider !== 'local') require('./vod-storage').deleteVodObjects(vod).catch(() => {}); } catch { /* */ }try { await db.run('DELETE FROM vods WHERE id = ?', [vodId]); } catch { /* */ }console.log(`[VOD] Discarded ephemeral clips-only recording (vod ${vodId})`);@@@ -247,22 +247,23 @@ async function _doFinalize(vodId, opts) {if (tools.getFileSizeSafe(filePath) === 0) {if (held) return await keepHeld('zero_byte');console.warn(`[VOD] vod ${vodId}: zero-byte recording — deleting empty recording`);-- try { fs.unlinkSync(filePath); } catch { /* */ }-- try { tools.cleanupSeekableFile(filePath); } catch { /* */ }-- try { if (vod.master_file_path && fs.existsSync(vod.master_file_path)) fs.unlinkSync(vod.master_file_path); } catch { /* */ }++ await tools.removeFile(filePath);++ await tools.cleanupSeekableFile(filePath);++ await tools.removeFile(vod.master_file_path);await _commitVodOutcome(vod, 'vod.failed', async () => await db.run('DELETE FROM vods WHERE id = ?', [vodId]),() => ({ ...vod, health_status: 'zero_byte', is_public: 0 }));return null;}++ // The live seekable copy is no longer needed: drop it (and stop a live pass still writing one)++ // before the final remux, whose own copy of the recording needs that room on the disk.++ await tools.cleanupSeekableFile(filePath);++// Remux for proper seeking support (fast copy-mode, no re-encode). Its last packet time is a// measurement of the file (source 'remux'); a failed remux measures nothing.const remux = await tools.remuxForSeekingDetailed(filePath);if (!remux.ok && remux.error !== 'unsupported container') issues.push('remux_failed');-- // Clean up the live seekable copy (no longer needed after final remux)-- tools.cleanupSeekableFile(filePath);--// The duration is what the file says: ffprobe's, else the packets' (a stream-copy pass when the// remux gave none), never the wall clock. Nothing measurable: 0 and needs_review.const measure = async (p, remuxS) => {@@@ -290,7 +291,7 @@ async function _doFinalize(vodId, opts) {const ok = await rebuildWebmFromMaster(masterPath, filePath);if (ok) {let again = { ok: false, seconds: 0 };-- try { again = await tools.remuxForSeekingDetailed(filePath); tools.cleanupSeekableFile(filePath); } catch { /* */ }++ try { again = await tools.remuxForSeekingDetailed(filePath); await tools.cleanupSeekableFile(filePath); } catch { /* */ }const rebuilt = await measure(filePath, again.ok ? again.seconds : 0);if (rebuilt.choice.seconds > durationSeconds) {({ probe, choice } = rebuilt);@@@ -354,7 +355,7 @@ async function _doFinalize(vodId, opts) {const webmComplete = !masterDur || durationSeconds >= masterDur - 8;if (fs.existsSync(masterPath)) {if (webmComplete) {-- fs.unlinkSync(masterPath);++ await fs.promises.unlink(masterPath);console.log(`[VOD] Removed master archive for vod ${vodId} (${path.basename(masterPath)})`);await db.run('UPDATE vods SET master_file_path = NULL WHERE id = ?', [vodId]); // not projected: no object change} else {@@@ -8,6 +8,7 @@const { spawn } = require('child_process');const fs = require('fs');++const tools = require('./media-tools');const db = require('../db/database');function probeMediaInfo(filePath) {@@@ -48,37 +49,11 @@ function decodeVodFile(filePath) {});}++// The shared remux (media-tools): a kill budget that scales with the file (this copy's flat 30 s cut++// every multi-GB recording short), a free-disk check, and its temp file removed off the event loop.++// → { ok, error }function remuxForSeeking(filePath) {-- return new Promise((resolve) => {-- // Match the tmp container to the source so an H.264 .mp4 VOD isn't-- // force-written into a VP8-only WebM (which just fails). For .mp4,-- // faststart makes it instantly seekable.-- const isMp4 = filePath.toLowerCase().endsWith('.mp4');-- const tmpPath = isMp4 ? `${filePath}.remux.tmp.mp4` : `${filePath}.remux.tmp.webm`;-- const args = isMp4-- ? ['-y', '-i', filePath, '-c', 'copy', '-fflags', '+genpts', '-movflags', '+faststart', tmpPath]-- : ['-y', '-i', filePath, '-c', 'copy', '-fflags', '+genpts', tmpPath];-- const proc = spawn('ffmpeg', args);-- proc.on('close', (code) => {-- if (code === 0 && fs.existsSync(tmpPath)) {-- try {-- fs.renameSync(tmpPath, filePath);-- resolve({ ok: true });-- } catch (err) {-- if (fs.existsSync(tmpPath)) fs.unlinkSync(tmpPath);-- resolve({ ok: false, error: err.message });-- }-- } else {-- if (fs.existsSync(tmpPath)) fs.unlinkSync(tmpPath);-- resolve({ ok: false, error: `ffmpeg remux failed code ${code}` });-- }-- });-- proc.on('error', (err) => {-- if (fs.existsSync(tmpPath)) fs.unlinkSync(tmpPath);-- resolve({ ok: false, error: err.message });-- });-- setTimeout(() => { try { proc.kill(); } catch {} }, 30000);-- });++ return tools.remuxForSeekingDetailed(filePath).then(({ ok, error }) => ({ ok, error }));}// Re-encode a served .webm from the lossless .master.mkv archive. Used by the@@@ -93,18 +68,18 @@ function _rebuildWebmFromMaster(masterPath, webmPath) {let ff;try { ff = spawn('ffmpeg', args, { stdio: 'ignore' }); } catch { return resolve(false); }const to = setTimeout(() => { try { ff.kill('SIGKILL'); } catch { /* */ } }, 45 * 60 * 1000);-- ff.on('close', (code) => {++ ff.on('close', async (code) => {clearTimeout(to);try {if (code === 0 && fs.existsSync(tmp) && fs.statSync(tmp).size > 1024) {-- fs.renameSync(tmp, webmPath);++ await fs.promises.rename(tmp, webmPath);return resolve(true);}} catch { /* fall through */ }-- try { fs.existsSync(tmp) && fs.unlinkSync(tmp); } catch { /* */ }++ await tools.removeFile(tmp);resolve(false);});-- ff.on('error', () => { clearTimeout(to); try { fs.existsSync(tmp) && fs.unlinkSync(tmp); } catch { /* */ } resolve(false); });++ ff.on('error', () => { clearTimeout(to); tools.removeFile(tmp).then(() => resolve(false)); });});}@@@ -9,6 +9,8 @@const path = require('path');const fs = require('fs');++const fsp = fs.promises;++const os = require('os');const { spawn } = require('child_process');/**@@@ -31,66 +33,101 @@ function tailAppend(buf, chunk, max = 16384) {return out.length > max ? out.slice(-max) : out;}++// ── Big files off the event loop ─────────────────────────────++// Freeing a multi-GB recording takes ext4 seconds under load, and an unlinkSync/renameSync holds++// every Media request for that long: on 2026-10-13 the live DVR remux of a 4-hour recording froze++// the event loop for up to 14 s at a time and Prometheus marked Media down. Recordings, their++// sidecars and remux temp files are removed and replaced with fs.promises only.++++/** Remove a file without holding the event loop. → true when nothing is left at that path; never throws. */++function removeFile(filePath) {++ if (!filePath) return Promise.resolve(false);++ return fsp.rm(filePath, { force: true }).then(() => true, () => false);++}++++/**++ * Whether the disk holding `dir` can take `bytes` more and still keep its reserve (5 GB or a tenth of++ * the disk, whichever is larger). A copy-mode remux writes a whole second copy of its input, and the++ * disk is shared with PostgreSQL. An unreadable statfs does not stop the caller.++ */++async function hasRoomFor(dir, bytes) {++ let s;++ try { s = await fsp.statfs(dir); } catch { return true; }++ const free = s.bavail * s.bsize;++ return free - bytes >= Math.max(5e9, s.blocks * s.bsize * 0.1);++}++++/**++ * Kill budget of a copy-mode remux of `bytes`: it still rewrites every byte (twice for +faststart),++ * and a flat 60 s killed every multi-hour recording, which left them fragmented with no seek index,++ * so every clip cut from cloud storage had to scan the whole file and timed out. ~45 s per GB,++ * 3 min floor, 45 min ceiling.++ */++function remuxBudgetMs(bytes) {++ return Math.min(45 * 60000, Math.max(180000, Math.round(bytes / 1e9 * 45000) + 60000));++}++++/**++ * Run ffmpeg to its end. → { code, stderr } (code null when it could not start or was killed).++ * SIGKILLed after budgetMs (onTimeout runs first); onSpawn sees the process (the live remux keeps it to stop it).++ */++function runFfmpeg(args, { budgetMs, onTimeout, onSpawn } = {}) {++ return new Promise((resolve) => {++ let proc;++ try { proc = spawn('ffmpeg', args, { stdio: ['ignore', 'ignore', 'pipe'] }); } catch (err) { return resolve({ code: null, stderr: err.message }); }++ let stderr = '', done = false;++ const killer = budgetMs ? setTimeout(() => { if (onTimeout) onTimeout(); try { proc.kill('SIGKILL'); } catch {} }, budgetMs) : null;++ const finish = (code) => { if (done) return; done = true; clearTimeout(killer); resolve({ code, stderr }); };++ proc.stderr.on('data', d => { stderr = tailAppend(stderr, String(d)); });++ proc.on('close', (code) => finish(code));++ proc.on('error', (err) => { stderr = tailAppend(stderr, err.message); finish(null); });++ if (onSpawn) onSpawn(proc);++ });++}++/*** Remux a WebM/MP4 file with ffmpeg to add proper seek metadata.* WebM files from MediaRecorder lack Cues and often have Inf duration,* which prevents browser-side seeking. A fast copy-remux fixes this.* MP4: rewrite the fragmented recording into a plain, faststart (moov-at-front)* MP4 so finished-VOD playback seeks instantly. Replaces the original in-place.++ * Skipped ('no room for the remux copy') when the disk cannot take the copy next to the original.** Resolves { ok, seconds, error }: `seconds` is the last packet time the remux wrote (0 when it* failed: a partial run's time says nothing about the file).*/--function remuxForSeekingDetailed(filePath) {-- return new Promise((resolve) => {-- const ext = path.extname(filePath).toLowerCase();-- if (ext !== '.webm' && ext !== '.mp4') return resolve({ ok: false, seconds: 0, error: 'unsupported container' }); // WebM + MP4 only---- const tmpPath = filePath + '.remux' + ext;-- const outArgs = ext === '.mp4'-- ? ['-c', 'copy', '-movflags', '+faststart', '-fflags', '+genpts']-- : ['-c', 'copy', '-fflags', '+genpts'];-- let proc;-- try {-- proc = spawn('ffmpeg', [-- '-y', '-i', filePath,-- ...outArgs,-- tmpPath,-- ], { stdio: ['ignore', 'ignore', 'pipe'] });-- } catch (err) { return resolve({ ok: false, seconds: 0, error: err.message }); }---- let stderr = '';-- proc.stderr.on('data', d => { stderr = tailAppend(stderr, String(d)); });---- proc.on('close', (code) => {-- if (code === 0 && fs.existsSync(tmpPath)) {-- try {-- fs.renameSync(tmpPath, filePath);-- console.log(`[VOD] Remuxed for seeking: ${path.basename(filePath)}`);-- resolve({ ok: true, seconds: lastProgressSeconds(stderr), error: null });-- } catch (err) {-- console.warn(`[VOD] Remux rename failed:`, err.message);-- try { fs.unlinkSync(tmpPath); } catch {}-- resolve({ ok: false, seconds: 0, error: err.message });-- }-- } else {-- console.warn(`[VOD] Remux failed (code ${code}): ${stderr.slice(-200)}`);-- try { if (fs.existsSync(tmpPath)) fs.unlinkSync(tmpPath); } catch {}-- resolve({ ok: false, seconds: 0, error: `ffmpeg remux exited ${code}` });-- }-- });---- proc.on('error', (err) => resolve({ ok: false, seconds: 0, error: err.message }));++async function remuxForSeekingDetailed(filePath) {++ const ext = path.extname(filePath).toLowerCase();++ if (ext !== '.webm' && ext !== '.mp4') return { ok: false, seconds: 0, error: 'unsupported container' }; // WebM + MP4 only++++ let bytes = 0; try { bytes = (await fsp.stat(filePath)).size; } catch { /* ffmpeg reports the missing file */ }++ if (!(await hasRoomFor(path.dirname(filePath), bytes))) {++ console.warn(`[VOD] Remux of ${path.basename(filePath)} skipped: not enough free disk for a ${(bytes / 1e9).toFixed(1)} GB copy`);++ return { ok: false, seconds: 0, error: 'no room for the remux copy' };++ }-- // Timeout scales with size: a copy-mode remux still rewrites every byte (twice for-- // +faststart), and a flat 60 s killed every multi-hour recording — which left them-- // fragmented with no seek index, so every clip cut from cloud storage had to scan-- // the whole file and timed out. ~45 s per GB, 3 min floor, 45 min ceiling.-- let bytes = 0; try { bytes = fs.statSync(filePath).size; } catch { /* */ }-- const budgetMs = Math.min(45 * 60000, Math.max(180000, Math.round(bytes / 1e9 * 45000) + 60000));-- const killer = setTimeout(() => { console.warn(`[VOD] Remux of ${path.basename(filePath)} exceeded ${Math.round(budgetMs / 1000)} s — killed`); try { proc.kill('SIGKILL'); } catch {} }, budgetMs);-- proc.on('close', () => clearTimeout(killer));++ const tmpPath = filePath + '.remux' + ext;++ const outArgs = ext === '.mp4'++ ? ['-c', 'copy', '-movflags', '+faststart', '-fflags', '+genpts']++ : ['-c', 'copy', '-fflags', '+genpts'];++ const budgetMs = remuxBudgetMs(bytes);++ const { code, stderr } = await runFfmpeg(['-y', '-i', filePath, ...outArgs, tmpPath], {++ budgetMs, onTimeout: () => console.warn(`[VOD] Remux of ${path.basename(filePath)} exceeded ${Math.round(budgetMs / 1000)} s — killed`),});++ if (code === 0 && fs.existsSync(tmpPath)) {++ try {++ await fsp.rename(tmpPath, filePath);++ console.log(`[VOD] Remuxed for seeking: ${path.basename(filePath)}`);++ return { ok: true, seconds: lastProgressSeconds(stderr), error: null };++ } catch (err) {++ console.warn(`[VOD] Remux rename failed:`, err.message);++ await removeFile(tmpPath);++ return { ok: false, seconds: 0, error: err.message };++ }++ }++ await removeFile(tmpPath);++ console.warn(`[VOD] Remux failed (code ${code}): ${stderr.slice(-200)}`);++ return { ok: false, seconds: 0, error: `ffmpeg remux exited ${code}` };}/** remuxForSeekingDetailed(), answering true/false (the callers that only need the outcome). */@@@ -253,7 +290,15 @@ function cleanupTempFile(filePath) {// ── Live DVR sidecar ─────────────────────────────────────────--const _liveRemuxInProgress = new Set();++// Per recording: { running, proc, notBefore, retired, warned }. Every pass rewrites the whole growing++// file (twice for MP4 +faststart), so a 4-hour recording's pass takes minutes and its copy is as big++// as the recording. One pass runs at a time, it may use at most 1/LIVE_REMUX_DUTY of the wall clock++// (the next waits 9× as long as the last took), its kill budget scales with the file, and it is++// skipped while the disk cannot take the copy. The old flat 30 s kill cut every long pass short,++// dropped the guard while ffmpeg still ran and let the next tick start another full copy.++const _liveRemux = new Map();++const LIVE_REMUX_DUTY = 10;++const LIVE_REMUX_NICE = 10;function seekableSidecarPath(filePath) {if (!filePath) return null;@@@ -267,67 +312,65 @@ function seekableSidecarPath(filePath) {* viewers can seek anywhere without the whole file. WebM: copy-remux to write* cues. MP4: copy-remux with +faststart so a complete moov sits at the front.* Writes to <file>.seekable.<ext> WITHOUT touching the growing original.++ * → true when a new snapshot replaced the old one; false when this call did nothing++ * (a pass is running or not yet due, no room, or the pass failed).*/--function remuxForLiveSeeking(filePath) {++async function remuxForLiveSeeking(filePath) {const seekablePath = seekableSidecarPath(filePath);-- if (!seekablePath) return Promise.resolve(false);-- if (_liveRemuxInProgress.has(filePath)) return Promise.resolve(false);-- _liveRemuxInProgress.add(filePath);---- const isMp4 = filePath.endsWith('.mp4');-- const tmpPath = filePath + (isMp4 ? '.live-remux.tmp.mp4' : '.live-remux.tmp.webm');-- const outArgs = isMp4-- ? ['-c', 'copy', '-movflags', '+faststart', '-fflags', '+genpts']-- : ['-c', 'copy', '-fflags', '+genpts'];---- return new Promise((resolve) => {-- const proc = spawn('ffmpeg', [-- '-y', '-i', filePath,-- ...outArgs,-- tmpPath,-- ], { stdio: ['ignore', 'ignore', 'pipe'] });---- let stderr = '';-- proc.stderr.on('data', d => stderr += d);---- proc.on('close', (code) => {-- _liveRemuxInProgress.delete(filePath);-- if (code === 0 && fs.existsSync(tmpPath)) {-- try {-- fs.renameSync(tmpPath, seekablePath);-- resolve(true);-- } catch (err) {-- console.warn(`[VOD] Live remux rename failed:`, err.message);-- try { fs.unlinkSync(tmpPath); } catch {}-- resolve(false);-- }-- } else {-- try { if (fs.existsSync(tmpPath)) fs.unlinkSync(tmpPath); } catch {}-- resolve(false);-- }-- });---- proc.on('error', () => {-- _liveRemuxInProgress.delete(filePath);-- resolve(false);++ if (!seekablePath) return false;++ let state = _liveRemux.get(filePath);++ if (!state) _liveRemux.set(filePath, state = { running: false, proc: null, notBefore: 0, retired: false, warned: false });++ if (state.running || Date.now() < state.notBefore) return false;++ state.running = true;++ const started = Date.now();++ const tmpPath = filePath + (filePath.endsWith('.mp4') ? '.live-remux.tmp.mp4' : '.live-remux.tmp.webm');++ try {++ let bytes;++ try { bytes = (await fsp.stat(filePath)).size; } catch { return false; }++ if (!(await hasRoomFor(path.dirname(filePath), bytes))) {++ if (!state.warned) console.warn(`[VOD] Live seek snapshot of ${path.basename(filePath)} paused: not enough free disk for a ${(bytes / 1e9).toFixed(1)} GB copy`);++ state.warned = true;++ return false;++ }++ state.warned = false;++ const outArgs = filePath.endsWith('.mp4')++ ? ['-c', 'copy', '-movflags', '+faststart', '-fflags', '+genpts']++ : ['-c', 'copy', '-fflags', '+genpts'];++ const { code } = await runFfmpeg(['-y', '-i', filePath, ...outArgs, tmpPath], {++ budgetMs: remuxBudgetMs(bytes),++ // Lower priority than the recordings it copies. Only with a pid: setPriority(undefined) renices Media itself.++ onSpawn: (proc) => { state.proc = proc; if (proc.pid) { try { os.setPriority(proc.pid, LIVE_REMUX_NICE); } catch { /* not permitted: run as is */ } } },});---- setTimeout(() => {-- _liveRemuxInProgress.delete(filePath);-- try { proc.kill(); } catch {}-- }, 30000);-- });++ state.proc = null;++ if (code === 0 && !state.retired && fs.existsSync(tmpPath)) {++ try {++ await fsp.rename(tmpPath, seekablePath);++ if (state.retired) await removeFile(seekablePath); // finalized while it was renaming++ return !state.retired;++ } catch (err) { console.warn(`[VOD] Live remux rename failed:`, err.message); }++ }++ await removeFile(tmpPath);++ return false;++ } finally {++ state.running = false;++ state.notBefore = Date.now() + (Date.now() - started) * (LIVE_REMUX_DUTY - 1);++ }}/*** Clean up the .seekable.* sidecar after finalization. Guard: the sidecar path* must differ from the VOD itself (a bad regex fall-through would otherwise* delete the recording); seekableSidecarPath returns null for unknown exts.++ * Stops a live pass still running for it (that pass removes its own temp file).++ * The removal runs off the event loop. → a promise of it; callers need not wait.*/function cleanupSeekableFile(filePath) {const seekablePath = seekableSidecarPath(filePath);-- if (!seekablePath || seekablePath === filePath) return;-- try { if (fs.existsSync(seekablePath)) fs.unlinkSync(seekablePath); } catch {}++ if (!seekablePath || seekablePath === filePath) return Promise.resolve(false);++ const state = _liveRemux.get(filePath);++ _liveRemux.delete(filePath);++ if (state) { state.retired = true; try { if (state.proc) state.proc.kill('SIGKILL'); } catch { /* already gone */ } }++ return removeFile(seekablePath);}// ── Browser-chunk segments (chunked upload flow) ─────────────@@@ -359,11 +402,10 @@ function getPendingSegmentFiles(filePath) {}}--function concatWebmFiles(basePath, appendPath) {-- if (!appendPath || !fs.existsSync(appendPath)) return Promise.resolve(true);++async function concatWebmFiles(basePath, appendPath) {++ if (!appendPath || !fs.existsSync(appendPath)) return true;if (!basePath || !fs.existsSync(basePath)) {-- fs.renameSync(appendPath, basePath);-- return Promise.resolve(true);++ try { await fsp.rename(appendPath, basePath); return true; } catch { return false; }}const listPath = `${basePath}.concat.${Date.now()}.txt`;@@@ -371,37 +413,24 @@ function concatWebmFiles(basePath, appendPath) {const escapedBase = basePath.replace(/'/g, `'\\''`);const escapedAppend = appendPath.replace(/'/g, `'\\''`);fs.writeFileSync(listPath, `file '${escapedBase}'\nfile '${escapedAppend}'\n`, 'utf8');++ // The copy pass rewrites the whole recording: its budget scales with it (a flat 120 s cut long ones short).++ const budgetMs = remuxBudgetMs(getFileSizeSafe(basePath) + getFileSizeSafe(appendPath));-- const runConcat = (args) => new Promise((resolve) => {-- const proc = spawn('ffmpeg', args, { stdio: ['ignore', 'ignore', 'pipe'] });-- let stderr = '';-- proc.stderr.on('data', (d) => stderr += d);-- proc.on('close', (code) => resolve({ code, stderr }));-- proc.on('error', () => resolve({ code: -1, stderr: 'spawn error' }));-- setTimeout(() => { try { proc.kill(); } catch {} }, 120000);-- });---- return (async () => {-- try {-- let result = await runConcat(['-y', '-f', 'concat', '-safe', '0', '-i', listPath, '-c', 'copy', '-fflags', '+genpts', tmpPath]);-- if (result.code !== 0 || !fs.existsSync(tmpPath)) {-- result = await runConcat(['-y', '-f', 'concat', '-safe', '0', '-i', listPath, '-c:v', 'libvpx-vp9', '-crf', '32', '-b:v', '0', '-c:a', 'libopus', '-b:a', '128k', tmpPath]);-- }-- if (result.code === 0 && fs.existsSync(tmpPath)) {-- fs.renameSync(tmpPath, basePath);-- cleanupTempFile(appendPath);-- cleanupTempFile(listPath);-- return true;-- }-- cleanupTempFile(tmpPath);-- cleanupTempFile(listPath);-- return false;-- } catch {-- cleanupTempFile(tmpPath);++ try {++ let result = await runFfmpeg(['-y', '-f', 'concat', '-safe', '0', '-i', listPath, '-c', 'copy', '-fflags', '+genpts', tmpPath], { budgetMs });++ if (result.code !== 0 || !fs.existsSync(tmpPath)) {++ result = await runFfmpeg(['-y', '-f', 'concat', '-safe', '0', '-i', listPath, '-c:v', 'libvpx-vp9', '-crf', '32', '-b:v', '0', '-c:a', 'libopus', '-b:a', '128k', tmpPath], { budgetMs });++ }++ if (result.code === 0 && fs.existsSync(tmpPath)) {++ await fsp.rename(tmpPath, basePath);++ await removeFile(appendPath);cleanupTempFile(listPath);-- return false;++ return true;}-- })();++ } catch { /* fall through: nothing merged */ }++ await removeFile(tmpPath);++ cleanupTempFile(listPath);++ return false;}/** Merge pending browser-chunk segments; resolves the number that could not be merged. */@@@ -432,6 +461,9 @@ module.exports = {probeVodInfo,getFileSizeSafe,cleanupTempFile,++ removeFile,++ hasRoomFor,++ remuxBudgetMs,makeSegmentPath,getPendingSegmentFiles,concatWebmFiles,@@@ -31,7 +31,6 @@const fs = require('fs');const path = require('path');--const { execSync } = require('child_process');const config = require('../config');const db = require('../db/database');@@@ -247,6 +246,7 @@ function endpointFor(name) {// placement (`write`) and re-projects the VOD's media_locations in one transaction (WS-G task 1;// afterTierMove). Lazy — the model requires this module.const objects = () => require('../objects/model');++const tools = () => require('./media-tools');async function _held(vod) { return await objects().isHeldRow(vod); }async function _moved(vodId, verified, write = null) { return await objects().afterTierMove(vodId, verified, write); }// After a move removed the local copy (the row already names the remote one): the object's locations@@@ -563,11 +563,10 @@ async function resolveMediaSource(vod) {// ── Tier moves ───────────────────────────────────────────────--function cleanupSidecar(localPath) {++/** Remove a VOD's .seekable.* sidecars, off the event loop (media-tools.removeFile). */++async function cleanupSidecar(localPath) {for (const seekable of [localPath.replace(/\.webm$/, '.seekable.webm'), localPath.replace(/\.mp4$/, '.seekable.mp4')]) {-- if (seekable !== localPath && fs.existsSync(seekable)) {-- try { fs.unlinkSync(seekable); } catch { /* ignore */ }-- }++ if (seekable !== localPath) await tools().removeFile(seekable);}}@@@ -602,10 +601,10 @@ async function moveToCold(vodId) {}await _moved(vodId, ['b2'], async () => await db.run("UPDATE vods SET storage_provider = 'b2', storage_key = ? WHERE id = ?", [key, vodId]));-- try { fs.unlinkSync(local); } catch (err) {++ try { await fs.promises.unlink(local); } catch (err) {console.error(`[VodStorage] Uploaded but failed to remove local file for VOD ${vodId}:`, err.message);}-- cleanupSidecar(local);++ await cleanupSidecar(local);await _localRemoved(vodId);console.log(`[VodStorage] VOD ${vodId} offloaded to B2: ${key} (${(localSize / 1048576).toFixed(1)} MB)`);@@@ -624,7 +623,7 @@ function cleanupStaleDownloads() {for (const f of fs.readdirSync(dir)) {if (!f.endsWith('.download')) continue;const fp = path.join(dir, f);-- try { if (Date.now() - fs.statSync(fp).mtimeMs > 3600000) { fs.unlinkSync(fp); console.log(`[VodStorage] Removed stale partial download ${f}`); } } catch { /* */ }++ try { if (Date.now() - fs.statSync(fp).mtimeMs > 3600000) tools().removeFile(fp).then((ok) => { if (ok) console.log(`[VodStorage] Removed stale partial download ${f}`); }); } catch { /* */ }}} catch { /* */ }}@@@ -696,12 +695,12 @@ async function _moveToHot(vodId) {obj.Body.on('error', finish);out.on('error', finish);out.on('finish', () => finish(null));-- }).catch((err) => { try { fs.unlinkSync(tmp); } catch { /* ignore */ } throw err; });++ }).catch(async (err) => { await tools().removeFile(tmp); throw err; });if (fs.statSync(tmp).size !== head.size) {-- try { fs.unlinkSync(tmp); } catch { /* ignore */ }++ await tools().removeFile(tmp);return { ok: false, error: 'Download verification failed' };}-- fs.renameSync(tmp, local);++ await fs.promises.rename(tmp, local);// Restoring to local keeps the B2 canonical copy; drop any R2 copy.if (providerOf(vod) === 'r2' && providerConfigured('r2')) await deleteObject('r2', key);@@@ -758,7 +757,7 @@ async function _promoteToR2(vodId) {let freed = 0;if (fs.existsSync(local)) {freed = fs.statSync(local).size;-- try { fs.unlinkSync(local); cleanupSidecar(local); } catch { freed = 0; }++ try { await fs.promises.unlink(local); await cleanupSidecar(local); } catch { freed = 0; }if (freed) await _localRemoved(vodId);}@@@ -809,14 +808,10 @@ async function deleteVodObjects(vod) {// Its CMAF segments and timeline rows (F3.1, objects/timeline.js) go with it.if (vod.object_id) await require('../objects/timeline').removeObject(vod.object_id).catch((err) => console.warn(`[VodStorage] timeline of ${vod.object_id}: ${err.message}`));const local = localPathForVod(vod);-- if (fs.existsSync(local)) {-- try { fs.unlinkSync(local); } catch { /* ignore */ }-- }++ await tools().removeFile(local);// Clips live in the clips dir — also try the record's own absolute path.-- if (vod.file_path !== local && fs.existsSync(vod.file_path)) {-- try { fs.unlinkSync(vod.file_path); } catch { /* ignore */ }-- }-- cleanupSidecar(local);++ if (vod.file_path !== local) await tools().removeFile(vod.file_path);++ await cleanupSidecar(local);const key = keyForVod(vod);for (const p of REMOTE_PROVIDERS) {if (providerConfigured(p)) await deleteObject(p, key);@@@ -825,20 +820,17 @@ async function deleteVodObjects(vod) {// ── Disk usage ───────────────────────────────────────────────++/**++ * The disk holding targetPath, as df reports it (usePct = used / (used + available), rounded up).++ * One statfs call: this used to run df through a shell with execSync, a process spawn on the event++ * loop for every sweep step and recorder check. `mount` is no longer known and stays '/'.++ */function diskUsage(targetPath) {try {-- const resolved = path.resolve(targetPath);-- const output = execSync(`df -B1 "${resolved}" 2>/dev/null | tail -1`, { encoding: 'utf8' });-- const parts = output.trim().split(/\s+/);-- if (parts.length >= 6) {-- return {-- total: parseInt(parts[1], 10) || 0,-- used: parseInt(parts[2], 10) || 0,-- available: parseInt(parts[3], 10) || 0,-- usePct: parseFloat(parts[4]) || 0,-- mount: parts[5] || '/',-- };-- }++ const s = fs.statfsSync(path.resolve(targetPath));++ const total = s.blocks * s.bsize, used = (s.blocks - s.bfree) * s.bsize, available = s.bavail * s.bsize;++ const usePct = used + available > 0 ? Math.ceil(used * 100 / (used + available)) : 0;++ return { total, used, available, usePct, mount: '/' };} catch { /* ignore */ }return { total: 0, used: 0, available: 0, usePct: 0, mount: '/' };}nnew file mode 100644@@@ -0,0 +1,129 @@++'use strict';++// The live DVR snapshot and the big-file deletes stay off the event loop. On 2026-10-13 a 4-hour++// recording (8 GB) froze Media for up to 14 s at a time and Prometheus marked it down: every 60 s the++// live remux copied the whole file, a flat 30 s kill cut each pass short and dropped its guard while++// ffmpeg still ran (so the next tick started another full copy), and the abandoned multi-GB temp file++// was freed with unlinkSync. Here ffmpeg is a stand-in process the test ends by hand.++const assert = require('assert');++const fs = require('fs');++const os = require('os');++const path = require('path');++const { EventEmitter } = require('events');++++const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-live-remux-'));++++// ffmpeg stand-in, installed before media-tools takes its reference to spawn.++const cp = require('child_process');++const spawned = [];++cp.spawn = (cmd, args) => {++ const p = new EventEmitter();++ p.stderr = new EventEmitter(); p.stdout = new EventEmitter(); p.pid = 999999999; // no such process: the renice of the pass fails quietly, and never touches the test p.killedWith = null;++ p.args = args;++ p.kill = (sig) => { p.killedWith = sig || 'SIGTERM'; setImmediate(() => p.emit('close', null, sig)); return true; };++ // "Finish" the pass: write its output (the last argument) and exit 0.++ p.finish = () => { fs.writeFileSync(args[args.length - 1], 'snapshot'); p.emit('close', 0, null); };++ spawned.push(p);++ return p;++};++++const tools = require('../server/vod/media-tools');++// Wait (up to 2 s) for a condition: a pass stats the file and the disk before it spawns.++const until = async (cond) => { for (let i = 0; i < 400 && !cond(); i++) await new Promise((res) => setTimeout(res, 5)); return cond(); };++// A roomy disk unless a section says otherwise (~100 GB, 41 GB free), so the result does not depend on the machine.++const realStatfs = fs.promises.statfs;++const roomy = async () => ({ bsize: 4096, blocks: 25e6, bfree: 10e6, bavail: 10e6 });++fs.promises.statfs = roomy;++++(async () => {++ const rec = path.join(tmp, 'vod-live-1-1.mp4');++ fs.writeFileSync(rec, Buffer.alloc(4096));++ const sidecar = tools.seekableSidecarPath(rec);++ const tmpOut = rec + '.live-remux.tmp.mp4';++++ // 1) One pass at a time: a second call while ffmpeg runs starts nothing.++ const first = tools.remuxForLiveSeeking(rec);++ assert.ok(await until(() => spawned.length === 1), 'the first call starts one pass');++ assert.strictEqual(await tools.remuxForLiveSeeking(rec), false, 'a call while the pass runs does nothing');++ assert.strictEqual(spawned.length, 1, 'no second ffmpeg on the same recording');++ assert.strictEqual(spawned[0].args[spawned[0].args.length - 1], tmpOut, 'the pass writes the temp file');++ assert.ok(spawned[0].args.includes('+faststart'), 'an MP4 snapshot is written with faststart');++++ // 2) The pass is not cut short at 30 s: its budget scales with the file, and it ends when ffmpeg ends.++ assert.strictEqual(tools.remuxBudgetMs(1e6), 180000, 'a small file gets the 3 min floor');++ assert.strictEqual(tools.remuxBudgetMs(8e9), 8 * 45000 + 60000, 'an 8 GB file gets ~45 s per GB');++ assert.strictEqual(tools.remuxBudgetMs(500e9), 45 * 60000, 'the ceiling is 45 min');++ spawned[0].finish();++ assert.strictEqual(await first, true, 'a finished pass replaces the snapshot');++ assert.strictEqual(fs.readFileSync(sidecar, 'utf8'), 'snapshot');++ assert.ok(!fs.existsSync(tmpOut), 'the temp file became the snapshot');++++ // 3) Paced: the next pass waits 9x as long as the last took, so an immediate call does nothing;++ // once that window has passed (however long the pass above took here), a new pass starts.++ assert.strictEqual(await tools.remuxForLiveSeeking(rec), false, 'a call inside the pacing window does nothing');++ assert.strictEqual(spawned.length, 1);++ let second = null;++ for (let i = 0; i < 200 && !second; i++) {++ await new Promise((res) => setTimeout(res, 50));++ const call = tools.remuxForLiveSeeking(rec); // not due: resolves false; due: runs until the stand-in ends++ const r = await Promise.race([call.then(() => 'returned'), until(() => spawned.length === 2).then((ok) => (ok ? 'spawned' : 'none'))]);++ if (r === 'spawned') second = call;++ }++ assert.ok(second && spawned.length === 2, 'after the pacing window a new pass starts');++++ // 4) Finalization retires the recording: the running pass is killed, nothing is left behind.++ fs.writeFileSync(tmpOut, 'partial');++ await tools.cleanupSeekableFile(rec);++ assert.strictEqual(spawned[1].killedWith, 'SIGKILL', 'the live pass is stopped at finalization');++ assert.strictEqual(await second, false);++ assert.ok(!fs.existsSync(sidecar), 'the snapshot is removed');++ assert.ok(!fs.existsSync(tmpOut), 'the killed pass removed its temp file');++++ // 5) No room: a pass that cannot fit its copy next to the recording is skipped (and starts no ffmpeg).++ fs.promises.statfs = async () => ({ bsize: 4096, blocks: 25e6, bfree: 1e5, bavail: 1e5 }); // ~100 GB disk, 0.4 GB free++ const big = path.join(tmp, 'vod-live-2-2.webm');++ fs.writeFileSync(big, Buffer.alloc(1024));++ const before = spawned.length;++ assert.strictEqual(await tools.remuxForLiveSeeking(big), false, 'no room: skipped');++ assert.strictEqual(spawned.length, before, 'no ffmpeg without room for its copy');++ const final = await tools.remuxForSeekingDetailed(big);++ assert.deepStrictEqual([final.ok, final.error], [false, 'no room for the remux copy'], 'the final remux checks the disk too');++ assert.strictEqual(spawned.length, before);++ assert.strictEqual(await tools.hasRoomFor(tmp, 0), false, '0.4 GB free is under the 10 GB reserve');++ fs.promises.statfs = roomy;++ assert.strictEqual(await tools.hasRoomFor(tmp, 20e9), true, '41 GB free takes a 20 GB copy and keeps 10 GB');++ assert.strictEqual(await tools.hasRoomFor(tmp, 35e9), false, 'but not a 35 GB one');++ fs.promises.statfs = realStatfs;++++ // 6) removeFile: off the event loop, never throws.++ const gone = path.join(tmp, 'gone.bin');++ fs.writeFileSync(gone, 'x');++ assert.strictEqual(await tools.removeFile(gone), true);++ assert.ok(!fs.existsSync(gone));++ assert.strictEqual(await tools.removeFile(gone), true, 'a missing file is already gone');++ assert.strictEqual(await tools.removeFile(null), false);++ const aDir = path.join(tmp, 'a-dir');++ fs.mkdirSync(aDir);++ assert.strictEqual(await tools.removeFile(aDir), false, 'a directory is not removed');++ assert.ok(fs.existsSync(aDir));++++ // 7) diskUsage reads statfs, not df through a shell.++ const vodStorage = require('../server/vod/vod-storage');++ const d = vodStorage.diskUsage(tmp);++ assert.ok(d.total > 0 && d.available > 0 && d.used >= 0 && d.usePct > 0 && d.usePct <= 100, `diskUsage: ${JSON.stringify(d)}`);++++ // 8) The VOD lifecycle files hold no synchronous unlink/rename of recordings, and no execSync.++ const src = (f) => fs.readFileSync(path.join(__dirname, '..', 'server', 'vod', f), 'utf8');++ for (const f of ['media-tools.js', 'finalize.js', 'vod-storage.js', 'health-scanner.js']) {++ const body = src(f).split('\n').filter((l) => !/^\s*(\/\/|\*)/.test(l)).join('\n');++ const hits = (body.match(/\b(renameSync|execSync)\b/g) || []);++ assert.deepStrictEqual(hits, [], `${f}: ${hits.join(', ')}`);++ }++ // cleanupTempFile (browser chunks) and the SDP files are small: the only unlinkSync left in these files.++ for (const f of ['finalize.js', 'vod-storage.js', 'health-scanner.js']) {++ assert.ok(!/unlinkSync/.test(src(f)), `${f} frees files with removeFile / fs.promises`);++ }++++ fs.rmSync(tmp, { recursive: true, force: true });++ console.log('live-remux-pacing: ok');++ process.exit(0);++})().catch((err) => { console.error(err); process.exit(1); });
@@ -103,18 +103,18 @@ function rebuildWebmFromMaster(masterPath, webmPath) {
@@ -103,18 +103,18 @@ function rebuildWebmFromMaster(masterPath, webmPath) {
let ff;
let ff;
try { ff = spawn('ffmpeg', args, { stdio: 'ignore' }); } catch { return resolve(false); }
try { ff = spawn('ffmpeg', args, { stdio: 'ignore' }); } catch { return resolve(false); }
const to = setTimeout(() => { try { ff.kill('SIGKILL'); } catch { /* */ } }, 45 * 60 * 1000);
const to = setTimeout(() => { try { ff.kill('SIGKILL'); } catch { /* */ } }, 45 * 60 * 1000);
- ff.on('close', (code) => {
+ ff.on('close', async (code) => {
clearTimeout(to);
clearTimeout(to);
try {
try {
if (code === 0 && fs.existsSync(tmp) && fs.statSync(tmp).size > 1024) {
if (code === 0 && fs.existsSync(tmp) && fs.statSync(tmp).size > 1024) {
- fs.renameSync(tmp, webmPath);
+ await fs.promises.rename(tmp, webmPath);
return resolve(true);
return resolve(true);
}
}
} catch { /* fall through */ }
} catch { /* fall through */ }
- try { fs.existsSync(tmp) && fs.unlinkSync(tmp); } catch { /* */ }
+ await tools.removeFile(tmp);
resolve(false);
resolve(false);
});
});
- ff.on('error', () => { clearTimeout(to); try { fs.existsSync(tmp) && fs.unlinkSync(tmp); } catch { /* */ } resolve(false); });
+ ff.on('error', () => { clearTimeout(to); tools.removeFile(tmp).then(() => resolve(false)); });
});
});
}
}
@@ -203,9 +203,9 @@ async function _doFinalize(vodId, opts) {
@@ -203,9 +203,9 @@ async function _doFinalize(vodId, opts) {
// Ephemeral clips-only recording: it existed only to serve live clips. Never
// Ephemeral clips-only recording: it existed only to serve live clips. Never
// published — delete the file (+ any offloaded object) and its row.
// published — delete the file (+ any offloaded object) and its row.
if (vod.clips_only) {
if (vod.clips_only) {
- try { if (filePath && fs.existsSync(filePath)) fs.unlinkSync(filePath); } catch { /* */ }
- try { tools.cleanupSeekableFile(filePath); } catch { /* */ }
- try { if (vod.master_file_path && fs.existsSync(vod.master_file_path)) fs.unlinkSync(vod.master_file_path); } catch { /* */ }
+ await tools.removeFile(filePath);
+ await tools.cleanupSeekableFile(filePath);
+ await tools.removeFile(vod.master_file_path);
try { if (vod.storage_provider && vod.storage_provider !== 'local') require('./vod-storage').deleteVodObjects(vod).catch(() => {}); } catch { /* */ }
try { if (vod.storage_provider && vod.storage_provider !== 'local') require('./vod-storage').deleteVodObjects(vod).catch(() => {}); } catch { /* */ }
try { await db.run('DELETE FROM vods WHERE id = ?', [vodId]); } catch { /* */ }
try { await db.run('DELETE FROM vods WHERE id = ?', [vodId]); } catch { /* */ }
console.log(`[VOD] Discarded ephemeral clips-only recording (vod ${vodId})`);
console.log(`[VOD] Discarded ephemeral clips-only recording (vod ${vodId})`);
@@ -247,22 +247,23 @@ async function _doFinalize(vodId, opts) {
@@ -247,22 +247,23 @@ async function _doFinalize(vodId, opts) {
if (tools.getFileSizeSafe(filePath) === 0) {
if (tools.getFileSizeSafe(filePath) === 0) {
if (held) return await keepHeld('zero_byte');
if (held) return await keepHeld('zero_byte');
console.warn(`[VOD] vod ${vodId}: zero-byte recording — deleting empty recording`);
console.warn(`[VOD] vod ${vodId}: zero-byte recording — deleting empty recording`);
- try { fs.unlinkSync(filePath); } catch { /* */ }
- try { tools.cleanupSeekableFile(filePath); } catch { /* */ }
- try { if (vod.master_file_path && fs.existsSync(vod.master_file_path)) fs.unlinkSync(vod.master_file_path); } catch { /* */ }
+ await tools.removeFile(filePath);
+ await tools.cleanupSeekableFile(filePath);
+ await tools.removeFile(vod.master_file_path);
await _commitVodOutcome(vod, 'vod.failed', async () => await db.run('DELETE FROM vods WHERE id = ?', [vodId]),
await _commitVodOutcome(vod, 'vod.failed', async () => await db.run('DELETE FROM vods WHERE id = ?', [vodId]),
() => ({ ...vod, health_status: 'zero_byte', is_public: 0 }));
() => ({ ...vod, health_status: 'zero_byte', is_public: 0 }));
return null;
return null;
}
}
+ // The live seekable copy is no longer needed: drop it (and stop a live pass still writing one)
+ // before the final remux, whose own copy of the recording needs that room on the disk.
+ await tools.cleanupSeekableFile(filePath);
+
// Remux for proper seeking support (fast copy-mode, no re-encode). Its last packet time is a
// Remux for proper seeking support (fast copy-mode, no re-encode). Its last packet time is a
// measurement of the file (source 'remux'); a failed remux measures nothing.
// measurement of the file (source 'remux'); a failed remux measures nothing.
const remux = await tools.remuxForSeekingDetailed(filePath);
const remux = await tools.remuxForSeekingDetailed(filePath);
if (!remux.ok && remux.error !== 'unsupported container') issues.push('remux_failed');
if (!remux.ok && remux.error !== 'unsupported container') issues.push('remux_failed');
- // Clean up the live seekable copy (no longer needed after final remux)
- tools.cleanupSeekableFile(filePath);
-
// The duration is what the file says: ffprobe's, else the packets' (a stream-copy pass when the
// The duration is what the file says: ffprobe's, else the packets' (a stream-copy pass when the
// remux gave none), never the wall clock. Nothing measurable: 0 and needs_review.
// remux gave none), never the wall clock. Nothing measurable: 0 and needs_review.
const measure = async (p, remuxS) => {
const measure = async (p, remuxS) => {
@@ -290,7 +291,7 @@ async function _doFinalize(vodId, opts) {
@@ -290,7 +291,7 @@ async function _doFinalize(vodId, opts) {
const ok = await rebuildWebmFromMaster(masterPath, filePath);
const ok = await rebuildWebmFromMaster(masterPath, filePath);
if (ok) {
if (ok) {
let again = { ok: false, seconds: 0 };
let again = { ok: false, seconds: 0 };
- try { again = await tools.remuxForSeekingDetailed(filePath); tools.cleanupSeekableFile(filePath); } catch { /* */ }
+ try { again = await tools.remuxForSeekingDetailed(filePath); await tools.cleanupSeekableFile(filePath); } catch { /* */ }
const rebuilt = await measure(filePath, again.ok ? again.seconds : 0);
const rebuilt = await measure(filePath, again.ok ? again.seconds : 0);
if (rebuilt.choice.seconds > durationSeconds) {
if (rebuilt.choice.seconds > durationSeconds) {
({ probe, choice } = rebuilt);
({ probe, choice } = rebuilt);
@@ -354,7 +355,7 @@ async function _doFinalize(vodId, opts) {
@@ -354,7 +355,7 @@ async function _doFinalize(vodId, opts) {
const webmComplete = !masterDur || durationSeconds >= masterDur - 8;
const webmComplete = !masterDur || durationSeconds >= masterDur - 8;
if (fs.existsSync(masterPath)) {
if (fs.existsSync(masterPath)) {
if (webmComplete) {
if (webmComplete) {
- fs.unlinkSync(masterPath);
+ await fs.promises.unlink(masterPath);
console.log(`[VOD] Removed master archive for vod ${vodId} (${path.basename(masterPath)})`);
console.log(`[VOD] Removed master archive for vod ${vodId} (${path.basename(masterPath)})`);
await db.run('UPDATE vods SET master_file_path = NULL WHERE id = ?', [vodId]); // not projected: no object change
await db.run('UPDATE vods SET master_file_path = NULL WHERE id = ?', [vodId]); // not projected: no object change
} else {
} else {
@@ -8,6 +8,7 @@
@@ -8,6 +8,7 @@
const { spawn } = require('child_process');
const { spawn } = require('child_process');
const fs = require('fs');
const fs = require('fs');
+const tools = require('./media-tools');
const db = require('../db/database');
const db = require('../db/database');
function probeMediaInfo(filePath) {
function probeMediaInfo(filePath) {
@@ -48,37 +49,11 @@ function decodeVodFile(filePath) {
@@ -48,37 +49,11 @@ function decodeVodFile(filePath) {
});
});
}
}
+// The shared remux (media-tools): a kill budget that scales with the file (this copy's flat 30 s cut
+// every multi-GB recording short), a free-disk check, and its temp file removed off the event loop.
+// → { ok, error }
function remuxForSeeking(filePath) {
function remuxForSeeking(filePath) {
- return new Promise((resolve) => {
- // Match the tmp container to the source so an H.264 .mp4 VOD isn't
- // force-written into a VP8-only WebM (which just fails). For .mp4,
- // faststart makes it instantly seekable.
- const isMp4 = filePath.toLowerCase().endsWith('.mp4');
- const tmpPath = isMp4 ? `${filePath}.remux.tmp.mp4` : `${filePath}.remux.tmp.webm`;
- const args = isMp4
- ? ['-y', '-i', filePath, '-c', 'copy', '-fflags', '+genpts', '-movflags', '+faststart', tmpPath]
- : ['-y', '-i', filePath, '-c', 'copy', '-fflags', '+genpts', tmpPath];
- const proc = spawn('ffmpeg', args);
- proc.on('close', (code) => {
- if (code === 0 && fs.existsSync(tmpPath)) {
- try {
- fs.renameSync(tmpPath, filePath);
- resolve({ ok: true });
- } catch (err) {
- if (fs.existsSync(tmpPath)) fs.unlinkSync(tmpPath);
- resolve({ ok: false, error: err.message });
- }
- } else {
- if (fs.existsSync(tmpPath)) fs.unlinkSync(tmpPath);
- resolve({ ok: false, error: `ffmpeg remux failed code ${code}` });
- }
- });
- proc.on('error', (err) => {
- if (fs.existsSync(tmpPath)) fs.unlinkSync(tmpPath);
- resolve({ ok: false, error: err.message });
- });
- setTimeout(() => { try { proc.kill(); } catch {} }, 30000);
- });
+ return tools.remuxForSeekingDetailed(filePath).then(({ ok, error }) => ({ ok, error }));
}
}
// Re-encode a served .webm from the lossless .master.mkv archive. Used by the
// Re-encode a served .webm from the lossless .master.mkv archive. Used by the
@@ -93,18 +68,18 @@ function _rebuildWebmFromMaster(masterPath, webmPath) {
@@ -93,18 +68,18 @@ function _rebuildWebmFromMaster(masterPath, webmPath) {
let ff;
let ff;
try { ff = spawn('ffmpeg', args, { stdio: 'ignore' }); } catch { return resolve(false); }
try { ff = spawn('ffmpeg', args, { stdio: 'ignore' }); } catch { return resolve(false); }
const to = setTimeout(() => { try { ff.kill('SIGKILL'); } catch { /* */ } }, 45 * 60 * 1000);
const to = setTimeout(() => { try { ff.kill('SIGKILL'); } catch { /* */ } }, 45 * 60 * 1000);
- ff.on('close', (code) => {
+ ff.on('close', async (code) => {
clearTimeout(to);
clearTimeout(to);
try {
try {
if (code === 0 && fs.existsSync(tmp) && fs.statSync(tmp).size > 1024) {
if (code === 0 && fs.existsSync(tmp) && fs.statSync(tmp).size > 1024) {
- fs.renameSync(tmp, webmPath);
+ await fs.promises.rename(tmp, webmPath);
return resolve(true);
return resolve(true);
}
}
} catch { /* fall through */ }
} catch { /* fall through */ }
- try { fs.existsSync(tmp) && fs.unlinkSync(tmp); } catch { /* */ }
+ await tools.removeFile(tmp);
resolve(false);
resolve(false);
});
});
- ff.on('error', () => { clearTimeout(to); try { fs.existsSync(tmp) && fs.unlinkSync(tmp); } catch { /* */ } resolve(false); });
+ ff.on('error', () => { clearTimeout(to); tools.removeFile(tmp).then(() => resolve(false)); });
});
});
}
}
@@ -9,6 +9,8 @@
@@ -9,6 +9,8 @@
const path = require('path');
const path = require('path');
const fs = require('fs');
const fs = require('fs');
+const fsp = fs.promises;
+const os = require('os');
const { spawn } = require('child_process');
const { spawn } = require('child_process');
/**
/**
@@ -31,66 +33,101 @@ function tailAppend(buf, chunk, max = 16384) {
@@ -31,66 +33,101 @@ function tailAppend(buf, chunk, max = 16384) {
return out.length > max ? out.slice(-max) : out;
return out.length > max ? out.slice(-max) : out;
}
}
+// ── Big files off the event loop ─────────────────────────────
+// Freeing a multi-GB recording takes ext4 seconds under load, and an unlinkSync/renameSync holds
+// every Media request for that long: on 2026-10-13 the live DVR remux of a 4-hour recording froze
+// the event loop for up to 14 s at a time and Prometheus marked Media down. Recordings, their
+// sidecars and remux temp files are removed and replaced with fs.promises only.
+
+/** Remove a file without holding the event loop. → true when nothing is left at that path; never throws. */
+function removeFile(filePath) {
+ if (!filePath) return Promise.resolve(false);
+ return fsp.rm(filePath, { force: true }).then(() => true, () => false);
+}
+
+/**
+ * Whether the disk holding `dir` can take `bytes` more and still keep its reserve (5 GB or a tenth of
+ * the disk, whichever is larger). A copy-mode remux writes a whole second copy of its input, and the
+ * disk is shared with PostgreSQL. An unreadable statfs does not stop the caller.
+ */
+async function hasRoomFor(dir, bytes) {
+ let s;
+ try { s = await fsp.statfs(dir); } catch { return true; }
+ const free = s.bavail * s.bsize;
+ return free - bytes >= Math.max(5e9, s.blocks * s.bsize * 0.1);
+}
+
+/**
+ * Kill budget of a copy-mode remux of `bytes`: it still rewrites every byte (twice for +faststart),
+ * and a flat 60 s killed every multi-hour recording, which left them fragmented with no seek index,
+ * so every clip cut from cloud storage had to scan the whole file and timed out. ~45 s per GB,
+ * 3 min floor, 45 min ceiling.
+ */
+function remuxBudgetMs(bytes) {
+ return Math.min(45 * 60000, Math.max(180000, Math.round(bytes / 1e9 * 45000) + 60000));
+}
+
+/**
+ * Run ffmpeg to its end. → { code, stderr } (code null when it could not start or was killed).
+ * SIGKILLed after budgetMs (onTimeout runs first); onSpawn sees the process (the live remux keeps it to stop it).
+ */
+function runFfmpeg(args, { budgetMs, onTimeout, onSpawn } = {}) {
+ return new Promise((resolve) => {
+ let proc;
+ try { proc = spawn('ffmpeg', args, { stdio: ['ignore', 'ignore', 'pipe'] }); } catch (err) { return resolve({ code: null, stderr: err.message }); }
+ let stderr = '', done = false;
+ const killer = budgetMs ? setTimeout(() => { if (onTimeout) onTimeout(); try { proc.kill('SIGKILL'); } catch {} }, budgetMs) : null;
+ const finish = (code) => { if (done) return; done = true; clearTimeout(killer); resolve({ code, stderr }); };
+ proc.stderr.on('data', d => { stderr = tailAppend(stderr, String(d)); });
+ proc.on('close', (code) => finish(code));
+ proc.on('error', (err) => { stderr = tailAppend(stderr, err.message); finish(null); });
+ if (onSpawn) onSpawn(proc);
+ });
+}
+
/**
/**
* Remux a WebM/MP4 file with ffmpeg to add proper seek metadata.
* Remux a WebM/MP4 file with ffmpeg to add proper seek metadata.
* WebM files from MediaRecorder lack Cues and often have Inf duration,
* WebM files from MediaRecorder lack Cues and often have Inf duration,
* which prevents browser-side seeking. A fast copy-remux fixes this.
* which prevents browser-side seeking. A fast copy-remux fixes this.
* MP4: rewrite the fragmented recording into a plain, faststart (moov-at-front)
* MP4: rewrite the fragmented recording into a plain, faststart (moov-at-front)
* MP4 so finished-VOD playback seeks instantly. Replaces the original in-place.
* MP4 so finished-VOD playback seeks instantly. Replaces the original in-place.
+ * Skipped ('no room for the remux copy') when the disk cannot take the copy next to the original.
*
*
* Resolves { ok, seconds, error }: `seconds` is the last packet time the remux wrote (0 when it
* Resolves { ok, seconds, error }: `seconds` is the last packet time the remux wrote (0 when it
* failed: a partial run's time says nothing about the file).
* failed: a partial run's time says nothing about the file).
*/
*/
-function remuxForSeekingDetailed(filePath) {
- return new Promise((resolve) => {
- const ext = path.extname(filePath).toLowerCase();
- if (ext !== '.webm' && ext !== '.mp4') return resolve({ ok: false, seconds: 0, error: 'unsupported container' }); // WebM + MP4 only
-
- const tmpPath = filePath + '.remux' + ext;
- const outArgs = ext === '.mp4'
- ? ['-c', 'copy', '-movflags', '+faststart', '-fflags', '+genpts']
- : ['-c', 'copy', '-fflags', '+genpts'];
- let proc;
- try {
- proc = spawn('ffmpeg', [
- '-y', '-i', filePath,
- ...outArgs,
- tmpPath,
- ], { stdio: ['ignore', 'ignore', 'pipe'] });
- } catch (err) { return resolve({ ok: false, seconds: 0, error: err.message }); }
-
- let stderr = '';
- proc.stderr.on('data', d => { stderr = tailAppend(stderr, String(d)); });
-
- proc.on('close', (code) => {
- if (code === 0 && fs.existsSync(tmpPath)) {
- try {
- fs.renameSync(tmpPath, filePath);
- console.log(`[VOD] Remuxed for seeking: ${path.basename(filePath)}`);
- resolve({ ok: true, seconds: lastProgressSeconds(stderr), error: null });
- } catch (err) {
- console.warn(`[VOD] Remux rename failed:`, err.message);
- try { fs.unlinkSync(tmpPath); } catch {}
- resolve({ ok: false, seconds: 0, error: err.message });
- }
- } else {
- console.warn(`[VOD] Remux failed (code ${code}): ${stderr.slice(-200)}`);
- try { if (fs.existsSync(tmpPath)) fs.unlinkSync(tmpPath); } catch {}
- resolve({ ok: false, seconds: 0, error: `ffmpeg remux exited ${code}` });
- }
- });
-
- proc.on('error', (err) => resolve({ ok: false, seconds: 0, error: err.message }));
+async function remuxForSeekingDetailed(filePath) {
+ const ext = path.extname(filePath).toLowerCase();
+ if (ext !== '.webm' && ext !== '.mp4') return { ok: false, seconds: 0, error: 'unsupported container' }; // WebM + MP4 only
+
+ let bytes = 0; try { bytes = (await fsp.stat(filePath)).size; } catch { /* ffmpeg reports the missing file */ }
+ if (!(await hasRoomFor(path.dirname(filePath), bytes))) {
+ console.warn(`[VOD] Remux of ${path.basename(filePath)} skipped: not enough free disk for a ${(bytes / 1e9).toFixed(1)} GB copy`);
+ return { ok: false, seconds: 0, error: 'no room for the remux copy' };
+ }
- // Timeout scales with size: a copy-mode remux still rewrites every byte (twice for
- // +faststart), and a flat 60 s killed every multi-hour recording — which left them
- // fragmented with no seek index, so every clip cut from cloud storage had to scan
- // the whole file and timed out. ~45 s per GB, 3 min floor, 45 min ceiling.
- let bytes = 0; try { bytes = fs.statSync(filePath).size; } catch { /* */ }
- const budgetMs = Math.min(45 * 60000, Math.max(180000, Math.round(bytes / 1e9 * 45000) + 60000));
- const killer = setTimeout(() => { console.warn(`[VOD] Remux of ${path.basename(filePath)} exceeded ${Math.round(budgetMs / 1000)} s — killed`); try { proc.kill('SIGKILL'); } catch {} }, budgetMs);
- proc.on('close', () => clearTimeout(killer));
+ const tmpPath = filePath + '.remux' + ext;
+ const outArgs = ext === '.mp4'
+ ? ['-c', 'copy', '-movflags', '+faststart', '-fflags', '+genpts']
+ : ['-c', 'copy', '-fflags', '+genpts'];
+ const budgetMs = remuxBudgetMs(bytes);
+ const { code, stderr } = await runFfmpeg(['-y', '-i', filePath, ...outArgs, tmpPath], {
+ budgetMs, onTimeout: () => console.warn(`[VOD] Remux of ${path.basename(filePath)} exceeded ${Math.round(budgetMs / 1000)} s — killed`),
});
});
+ if (code === 0 && fs.existsSync(tmpPath)) {
+ try {
+ await fsp.rename(tmpPath, filePath);
+ console.log(`[VOD] Remuxed for seeking: ${path.basename(filePath)}`);
+ return { ok: true, seconds: lastProgressSeconds(stderr), error: null };
+ } catch (err) {
+ console.warn(`[VOD] Remux rename failed:`, err.message);
+ await removeFile(tmpPath);
+ return { ok: false, seconds: 0, error: err.message };
+ }
+ }
+ await removeFile(tmpPath);
+ console.warn(`[VOD] Remux failed (code ${code}): ${stderr.slice(-200)}`);
+ return { ok: false, seconds: 0, error: `ffmpeg remux exited ${code}` };
}
}
/** remuxForSeekingDetailed(), answering true/false (the callers that only need the outcome). */
/** remuxForSeekingDetailed(), answering true/false (the callers that only need the outcome). */
@@ -253,7 +290,15 @@ function cleanupTempFile(filePath) {
@@ -253,7 +290,15 @@ function cleanupTempFile(filePath) {
// ── Live DVR sidecar ─────────────────────────────────────────
// ── Live DVR sidecar ─────────────────────────────────────────
-const _liveRemuxInProgress = new Set();
+// Per recording: { running, proc, notBefore, retired, warned }. Every pass rewrites the whole growing
+// file (twice for MP4 +faststart), so a 4-hour recording's pass takes minutes and its copy is as big
+// as the recording. One pass runs at a time, it may use at most 1/LIVE_REMUX_DUTY of the wall clock
+// (the next waits 9× as long as the last took), its kill budget scales with the file, and it is
+// skipped while the disk cannot take the copy. The old flat 30 s kill cut every long pass short,
+// dropped the guard while ffmpeg still ran and let the next tick start another full copy.
+const _liveRemux = new Map();
+const LIVE_REMUX_DUTY = 10;
+const LIVE_REMUX_NICE = 10;
function seekableSidecarPath(filePath) {
function seekableSidecarPath(filePath) {
if (!filePath) return null;
if (!filePath) return null;
@@ -267,67 +312,65 @@ function seekableSidecarPath(filePath) {
@@ -267,67 +312,65 @@ function seekableSidecarPath(filePath) {
* viewers can seek anywhere without the whole file. WebM: copy-remux to write
* viewers can seek anywhere without the whole file. WebM: copy-remux to write
* cues. MP4: copy-remux with +faststart so a complete moov sits at the front.
* cues. MP4: copy-remux with +faststart so a complete moov sits at the front.
* Writes to <file>.seekable.<ext> WITHOUT touching the growing original.
* Writes to <file>.seekable.<ext> WITHOUT touching the growing original.
+ * → true when a new snapshot replaced the old one; false when this call did nothing
+ * (a pass is running or not yet due, no room, or the pass failed).
*/
*/
-function remuxForLiveSeeking(filePath) {
+async function remuxForLiveSeeking(filePath) {
const seekablePath = seekableSidecarPath(filePath);
const seekablePath = seekableSidecarPath(filePath);
- if (!seekablePath) return Promise.resolve(false);
- if (_liveRemuxInProgress.has(filePath)) return Promise.resolve(false);
- _liveRemuxInProgress.add(filePath);
-
- const isMp4 = filePath.endsWith('.mp4');
- const tmpPath = filePath + (isMp4 ? '.live-remux.tmp.mp4' : '.live-remux.tmp.webm');
- const outArgs = isMp4
- ? ['-c', 'copy', '-movflags', '+faststart', '-fflags', '+genpts']
- : ['-c', 'copy', '-fflags', '+genpts'];
-
- return new Promise((resolve) => {
- const proc = spawn('ffmpeg', [
- '-y', '-i', filePath,
- ...outArgs,
- tmpPath,
- ], { stdio: ['ignore', 'ignore', 'pipe'] });
-
- let stderr = '';
- proc.stderr.on('data', d => stderr += d);
-
- proc.on('close', (code) => {
- _liveRemuxInProgress.delete(filePath);
- if (code === 0 && fs.existsSync(tmpPath)) {
- try {
- fs.renameSync(tmpPath, seekablePath);
- resolve(true);
- } catch (err) {
- console.warn(`[VOD] Live remux rename failed:`, err.message);
- try { fs.unlinkSync(tmpPath); } catch {}
- resolve(false);
- }
- } else {
- try { if (fs.existsSync(tmpPath)) fs.unlinkSync(tmpPath); } catch {}
- resolve(false);
- }
- });
-
- proc.on('error', () => {
- _liveRemuxInProgress.delete(filePath);
- resolve(false);
+ if (!seekablePath) return false;
+ let state = _liveRemux.get(filePath);
+ if (!state) _liveRemux.set(filePath, state = { running: false, proc: null, notBefore: 0, retired: false, warned: false });
+ if (state.running || Date.now() < state.notBefore) return false;
+ state.running = true;
+ const started = Date.now();
+ const tmpPath = filePath + (filePath.endsWith('.mp4') ? '.live-remux.tmp.mp4' : '.live-remux.tmp.webm');
+ try {
+ let bytes;
+ try { bytes = (await fsp.stat(filePath)).size; } catch { return false; }
+ if (!(await hasRoomFor(path.dirname(filePath), bytes))) {
+ if (!state.warned) console.warn(`[VOD] Live seek snapshot of ${path.basename(filePath)} paused: not enough free disk for a ${(bytes / 1e9).toFixed(1)} GB copy`);
+ state.warned = true;
+ return false;
+ }
+ state.warned = false;
+ const outArgs = filePath.endsWith('.mp4')
+ ? ['-c', 'copy', '-movflags', '+faststart', '-fflags', '+genpts']
+ : ['-c', 'copy', '-fflags', '+genpts'];
+ const { code } = await runFfmpeg(['-y', '-i', filePath, ...outArgs, tmpPath], {
+ budgetMs: remuxBudgetMs(bytes),
+ // Lower priority than the recordings it copies. Only with a pid: setPriority(undefined) renices Media itself.
+ onSpawn: (proc) => { state.proc = proc; if (proc.pid) { try { os.setPriority(proc.pid, LIVE_REMUX_NICE); } catch { /* not permitted: run as is */ } } },
});
});
-
- setTimeout(() => {
- _liveRemuxInProgress.delete(filePath);
- try { proc.kill(); } catch {}
- }, 30000);
- });
+ state.proc = null;
+ if (code === 0 && !state.retired && fs.existsSync(tmpPath)) {
+ try {
+ await fsp.rename(tmpPath, seekablePath);
+ if (state.retired) await removeFile(seekablePath); // finalized while it was renaming
+ return !state.retired;
+ } catch (err) { console.warn(`[VOD] Live remux rename failed:`, err.message); }
+ }
+ await removeFile(tmpPath);
+ return false;
+ } finally {
+ state.running = false;
+ state.notBefore = Date.now() + (Date.now() - started) * (LIVE_REMUX_DUTY - 1);
+ }
}
}
/**
/**
* Clean up the .seekable.* sidecar after finalization. Guard: the sidecar path
* Clean up the .seekable.* sidecar after finalization. Guard: the sidecar path
* must differ from the VOD itself (a bad regex fall-through would otherwise
* must differ from the VOD itself (a bad regex fall-through would otherwise
* delete the recording); seekableSidecarPath returns null for unknown exts.
* delete the recording); seekableSidecarPath returns null for unknown exts.
+ * Stops a live pass still running for it (that pass removes its own temp file).
+ * The removal runs off the event loop. → a promise of it; callers need not wait.
*/
*/
function cleanupSeekableFile(filePath) {
function cleanupSeekableFile(filePath) {
const seekablePath = seekableSidecarPath(filePath);
const seekablePath = seekableSidecarPath(filePath);
- if (!seekablePath || seekablePath === filePath) return;
- try { if (fs.existsSync(seekablePath)) fs.unlinkSync(seekablePath); } catch {}
+ if (!seekablePath || seekablePath === filePath) return Promise.resolve(false);
+ const state = _liveRemux.get(filePath);
+ _liveRemux.delete(filePath);
+ if (state) { state.retired = true; try { if (state.proc) state.proc.kill('SIGKILL'); } catch { /* already gone */ } }
+ return removeFile(seekablePath);
}
}
// ── Browser-chunk segments (chunked upload flow) ─────────────
// ── Browser-chunk segments (chunked upload flow) ─────────────
@@ -359,11 +402,10 @@ function getPendingSegmentFiles(filePath) {
@@ -359,11 +402,10 @@ function getPendingSegmentFiles(filePath) {
}
}
}
}
-function concatWebmFiles(basePath, appendPath) {
- if (!appendPath || !fs.existsSync(appendPath)) return Promise.resolve(true);
+async function concatWebmFiles(basePath, appendPath) {
+ if (!appendPath || !fs.existsSync(appendPath)) return true;
if (!basePath || !fs.existsSync(basePath)) {
if (!basePath || !fs.existsSync(basePath)) {
- fs.renameSync(appendPath, basePath);
- return Promise.resolve(true);
+ try { await fsp.rename(appendPath, basePath); return true; } catch { return false; }
}
}
const listPath = `${basePath}.concat.${Date.now()}.txt`;
const listPath = `${basePath}.concat.${Date.now()}.txt`;
@@ -371,37 +413,24 @@ function concatWebmFiles(basePath, appendPath) {
@@ -371,37 +413,24 @@ function concatWebmFiles(basePath, appendPath) {
const escapedBase = basePath.replace(/'/g, `'\\''`);
const escapedBase = basePath.replace(/'/g, `'\\''`);
const escapedAppend = appendPath.replace(/'/g, `'\\''`);
const escapedAppend = appendPath.replace(/'/g, `'\\''`);
fs.writeFileSync(listPath, `file '${escapedBase}'\nfile '${escapedAppend}'\n`, 'utf8');
fs.writeFileSync(listPath, `file '${escapedBase}'\nfile '${escapedAppend}'\n`, 'utf8');
+ // The copy pass rewrites the whole recording: its budget scales with it (a flat 120 s cut long ones short).
+ const budgetMs = remuxBudgetMs(getFileSizeSafe(basePath) + getFileSizeSafe(appendPath));
- const runConcat = (args) => new Promise((resolve) => {
- const proc = spawn('ffmpeg', args, { stdio: ['ignore', 'ignore', 'pipe'] });
- let stderr = '';
- proc.stderr.on('data', (d) => stderr += d);
- proc.on('close', (code) => resolve({ code, stderr }));
- proc.on('error', () => resolve({ code: -1, stderr: 'spawn error' }));
- setTimeout(() => { try { proc.kill(); } catch {} }, 120000);
- });
-
- return (async () => {
- try {
- let result = await runConcat(['-y', '-f', 'concat', '-safe', '0', '-i', listPath, '-c', 'copy', '-fflags', '+genpts', tmpPath]);
- if (result.code !== 0 || !fs.existsSync(tmpPath)) {
- result = await runConcat(['-y', '-f', 'concat', '-safe', '0', '-i', listPath, '-c:v', 'libvpx-vp9', '-crf', '32', '-b:v', '0', '-c:a', 'libopus', '-b:a', '128k', tmpPath]);
- }
- if (result.code === 0 && fs.existsSync(tmpPath)) {
- fs.renameSync(tmpPath, basePath);
- cleanupTempFile(appendPath);
- cleanupTempFile(listPath);
- return true;
- }
- cleanupTempFile(tmpPath);
- cleanupTempFile(listPath);
- return false;
- } catch {
- cleanupTempFile(tmpPath);
+ try {
+ let result = await runFfmpeg(['-y', '-f', 'concat', '-safe', '0', '-i', listPath, '-c', 'copy', '-fflags', '+genpts', tmpPath], { budgetMs });
+ if (result.code !== 0 || !fs.existsSync(tmpPath)) {
+ result = await runFfmpeg(['-y', '-f', 'concat', '-safe', '0', '-i', listPath, '-c:v', 'libvpx-vp9', '-crf', '32', '-b:v', '0', '-c:a', 'libopus', '-b:a', '128k', tmpPath], { budgetMs });
+ }
+ if (result.code === 0 && fs.existsSync(tmpPath)) {
+ await fsp.rename(tmpPath, basePath);
+ await removeFile(appendPath);
cleanupTempFile(listPath);
cleanupTempFile(listPath);
- return false;
+ return true;
}
}
- })();
+ } catch { /* fall through: nothing merged */ }
+ await removeFile(tmpPath);
+ cleanupTempFile(listPath);
+ return false;
}
}
/** Merge pending browser-chunk segments; resolves the number that could not be merged. */
/** Merge pending browser-chunk segments; resolves the number that could not be merged. */
@@ -432,6 +461,9 @@ module.exports = {
@@ -432,6 +461,9 @@ module.exports = {
probeVodInfo,
probeVodInfo,
getFileSizeSafe,
getFileSizeSafe,
cleanupTempFile,
cleanupTempFile,
+ removeFile,
+ hasRoomFor,
+ remuxBudgetMs,
makeSegmentPath,
makeSegmentPath,
getPendingSegmentFiles,
getPendingSegmentFiles,
concatWebmFiles,
concatWebmFiles,
@@ -31,7 +31,6 @@
@@ -31,7 +31,6 @@
const fs = require('fs');
const fs = require('fs');
const path = require('path');
const path = require('path');
-const { execSync } = require('child_process');
const config = require('../config');
const config = require('../config');
const db = require('../db/database');
const db = require('../db/database');
@@ -247,6 +246,7 @@ function endpointFor(name) {
@@ -247,6 +246,7 @@ function endpointFor(name) {
// placement (`write`) and re-projects the VOD's media_locations in one transaction (WS-G task 1;
// placement (`write`) and re-projects the VOD's media_locations in one transaction (WS-G task 1;
// afterTierMove). Lazy — the model requires this module.
// afterTierMove). Lazy — the model requires this module.
const objects = () => require('../objects/model');
const objects = () => require('../objects/model');
+const tools = () => require('./media-tools');
async function _held(vod) { return await objects().isHeldRow(vod); }
async function _held(vod) { return await objects().isHeldRow(vod); }
async function _moved(vodId, verified, write = null) { return await objects().afterTierMove(vodId, verified, write); }
async function _moved(vodId, verified, write = null) { return await objects().afterTierMove(vodId, verified, write); }
// After a move removed the local copy (the row already names the remote one): the object's locations
// After a move removed the local copy (the row already names the remote one): the object's locations
@@ -563,11 +563,10 @@ async function resolveMediaSource(vod) {
@@ -563,11 +563,10 @@ async function resolveMediaSource(vod) {
// ── Tier moves ───────────────────────────────────────────────
// ── Tier moves ───────────────────────────────────────────────
-function cleanupSidecar(localPath) {
+/** Remove a VOD's .seekable.* sidecars, off the event loop (media-tools.removeFile). */
+async function cleanupSidecar(localPath) {
for (const seekable of [localPath.replace(/\.webm$/, '.seekable.webm'), localPath.replace(/\.mp4$/, '.seekable.mp4')]) {
for (const seekable of [localPath.replace(/\.webm$/, '.seekable.webm'), localPath.replace(/\.mp4$/, '.seekable.mp4')]) {
- if (seekable !== localPath && fs.existsSync(seekable)) {
- try { fs.unlinkSync(seekable); } catch { /* ignore */ }
- }
+ if (seekable !== localPath) await tools().removeFile(seekable);
}
}
}
}
@@ -602,10 +601,10 @@ async function moveToCold(vodId) {
@@ -602,10 +601,10 @@ async function moveToCold(vodId) {
}
}
await _moved(vodId, ['b2'], async () => await db.run("UPDATE vods SET storage_provider = 'b2', storage_key = ? WHERE id = ?", [key, vodId]));
await _moved(vodId, ['b2'], async () => await db.run("UPDATE vods SET storage_provider = 'b2', storage_key = ? WHERE id = ?", [key, vodId]));
- try { fs.unlinkSync(local); } catch (err) {
+ try { await fs.promises.unlink(local); } catch (err) {
console.error(`[VodStorage] Uploaded but failed to remove local file for VOD ${vodId}:`, err.message);
console.error(`[VodStorage] Uploaded but failed to remove local file for VOD ${vodId}:`, err.message);
}
}
- cleanupSidecar(local);
+ await cleanupSidecar(local);
await _localRemoved(vodId);
await _localRemoved(vodId);
console.log(`[VodStorage] VOD ${vodId} offloaded to B2: ${key} (${(localSize / 1048576).toFixed(1)} MB)`);
console.log(`[VodStorage] VOD ${vodId} offloaded to B2: ${key} (${(localSize / 1048576).toFixed(1)} MB)`);
@@ -624,7 +623,7 @@ function cleanupStaleDownloads() {
@@ -624,7 +623,7 @@ function cleanupStaleDownloads() {
for (const f of fs.readdirSync(dir)) {
for (const f of fs.readdirSync(dir)) {
if (!f.endsWith('.download')) continue;
if (!f.endsWith('.download')) continue;
const fp = path.join(dir, f);
const fp = path.join(dir, f);
- try { if (Date.now() - fs.statSync(fp).mtimeMs > 3600000) { fs.unlinkSync(fp); console.log(`[VodStorage] Removed stale partial download ${f}`); } } catch { /* */ }
+ try { if (Date.now() - fs.statSync(fp).mtimeMs > 3600000) tools().removeFile(fp).then((ok) => { if (ok) console.log(`[VodStorage] Removed stale partial download ${f}`); }); } catch { /* */ }
}
}
} catch { /* */ }
} catch { /* */ }
}
}
@@ -696,12 +695,12 @@ async function _moveToHot(vodId) {
@@ -696,12 +695,12 @@ async function _moveToHot(vodId) {
obj.Body.on('error', finish);
obj.Body.on('error', finish);
out.on('error', finish);
out.on('error', finish);
out.on('finish', () => finish(null));
out.on('finish', () => finish(null));
- }).catch((err) => { try { fs.unlinkSync(tmp); } catch { /* ignore */ } throw err; });
+ }).catch(async (err) => { await tools().removeFile(tmp); throw err; });
if (fs.statSync(tmp).size !== head.size) {
if (fs.statSync(tmp).size !== head.size) {
- try { fs.unlinkSync(tmp); } catch { /* ignore */ }
+ await tools().removeFile(tmp);
return { ok: false, error: 'Download verification failed' };
return { ok: false, error: 'Download verification failed' };
}
}
- fs.renameSync(tmp, local);
+ await fs.promises.rename(tmp, local);
// Restoring to local keeps the B2 canonical copy; drop any R2 copy.
// Restoring to local keeps the B2 canonical copy; drop any R2 copy.
if (providerOf(vod) === 'r2' && providerConfigured('r2')) await deleteObject('r2', key);
if (providerOf(vod) === 'r2' && providerConfigured('r2')) await deleteObject('r2', key);
@@ -758,7 +757,7 @@ async function _promoteToR2(vodId) {
@@ -758,7 +757,7 @@ async function _promoteToR2(vodId) {
let freed = 0;
let freed = 0;
if (fs.existsSync(local)) {
if (fs.existsSync(local)) {
freed = fs.statSync(local).size;
freed = fs.statSync(local).size;
- try { fs.unlinkSync(local); cleanupSidecar(local); } catch { freed = 0; }
+ try { await fs.promises.unlink(local); await cleanupSidecar(local); } catch { freed = 0; }
if (freed) await _localRemoved(vodId);
if (freed) await _localRemoved(vodId);
}
}
@@ -809,14 +808,10 @@ async function deleteVodObjects(vod) {
@@ -809,14 +808,10 @@ async function deleteVodObjects(vod) {
// Its CMAF segments and timeline rows (F3.1, objects/timeline.js) go with it.
// Its CMAF segments and timeline rows (F3.1, objects/timeline.js) go with it.
if (vod.object_id) await require('../objects/timeline').removeObject(vod.object_id).catch((err) => console.warn(`[VodStorage] timeline of ${vod.object_id}: ${err.message}`));
if (vod.object_id) await require('../objects/timeline').removeObject(vod.object_id).catch((err) => console.warn(`[VodStorage] timeline of ${vod.object_id}: ${err.message}`));
const local = localPathForVod(vod);
const local = localPathForVod(vod);
- if (fs.existsSync(local)) {
- try { fs.unlinkSync(local); } catch { /* ignore */ }
- }
+ await tools().removeFile(local);
// Clips live in the clips dir — also try the record's own absolute path.
// Clips live in the clips dir — also try the record's own absolute path.
- if (vod.file_path !== local && fs.existsSync(vod.file_path)) {
- try { fs.unlinkSync(vod.file_path); } catch { /* ignore */ }
- }
- cleanupSidecar(local);
+ if (vod.file_path !== local) await tools().removeFile(vod.file_path);
+ await cleanupSidecar(local);
const key = keyForVod(vod);
const key = keyForVod(vod);
for (const p of REMOTE_PROVIDERS) {
for (const p of REMOTE_PROVIDERS) {
if (providerConfigured(p)) await deleteObject(p, key);
if (providerConfigured(p)) await deleteObject(p, key);
@@ -825,20 +820,17 @@ async function deleteVodObjects(vod) {
@@ -825,20 +820,17 @@ async function deleteVodObjects(vod) {
// ── Disk usage ───────────────────────────────────────────────
// ── Disk usage ───────────────────────────────────────────────
+/**
+ * The disk holding targetPath, as df reports it (usePct = used / (used + available), rounded up).
+ * One statfs call: this used to run df through a shell with execSync, a process spawn on the event
+ * loop for every sweep step and recorder check. `mount` is no longer known and stays '/'.
+ */
function diskUsage(targetPath) {
function diskUsage(targetPath) {
try {
try {
- const resolved = path.resolve(targetPath);
- const output = execSync(`df -B1 "${resolved}" 2>/dev/null | tail -1`, { encoding: 'utf8' });
- const parts = output.trim().split(/\s+/);
- if (parts.length >= 6) {
- return {
- total: parseInt(parts[1], 10) || 0,
- used: parseInt(parts[2], 10) || 0,
- available: parseInt(parts[3], 10) || 0,
- usePct: parseFloat(parts[4]) || 0,
- mount: parts[5] || '/',
- };
- }
+ const s = fs.statfsSync(path.resolve(targetPath));
+ const total = s.blocks * s.bsize, used = (s.blocks - s.bfree) * s.bsize, available = s.bavail * s.bsize;
+ const usePct = used + available > 0 ? Math.ceil(used * 100 / (used + available)) : 0;
+ return { total, used, available, usePct, mount: '/' };
} catch { /* ignore */ }
} catch { /* ignore */ }
return { total: 0, used: 0, available: 0, usePct: 0, mount: '/' };
return { total: 0, used: 0, available: 0, usePct: 0, mount: '/' };
}
}
new file mode 100644
new file mode 100644
@@ -0,0 +1,129 @@
@@ -0,0 +1,129 @@
+'use strict';
+// The live DVR snapshot and the big-file deletes stay off the event loop. On 2026-10-13 a 4-hour
+// recording (8 GB) froze Media for up to 14 s at a time and Prometheus marked it down: every 60 s the
+// live remux copied the whole file, a flat 30 s kill cut each pass short and dropped its guard while
+// ffmpeg still ran (so the next tick started another full copy), and the abandoned multi-GB temp file
+// was freed with unlinkSync. Here ffmpeg is a stand-in process the test ends by hand.
+const assert = require('assert');
+const fs = require('fs');
+const os = require('os');
+const path = require('path');
+const { EventEmitter } = require('events');
+
+const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-live-remux-'));
+
+// ffmpeg stand-in, installed before media-tools takes its reference to spawn.
+const cp = require('child_process');
+const spawned = [];
+cp.spawn = (cmd, args) => {
+ const p = new EventEmitter();
+ p.stderr = new EventEmitter(); p.stdout = new EventEmitter(); p.pid = 999999999; // no such process: the renice of the pass fails quietly, and never touches the test p.killedWith = null;
+ p.args = args;
+ p.kill = (sig) => { p.killedWith = sig || 'SIGTERM'; setImmediate(() => p.emit('close', null, sig)); return true; };
+ // "Finish" the pass: write its output (the last argument) and exit 0.
+ p.finish = () => { fs.writeFileSync(args[args.length - 1], 'snapshot'); p.emit('close', 0, null); };
+ spawned.push(p);
+ return p;
+};
+
+const tools = require('../server/vod/media-tools');
+// Wait (up to 2 s) for a condition: a pass stats the file and the disk before it spawns.
+const until = async (cond) => { for (let i = 0; i < 400 && !cond(); i++) await new Promise((res) => setTimeout(res, 5)); return cond(); };
+// A roomy disk unless a section says otherwise (~100 GB, 41 GB free), so the result does not depend on the machine.
+const realStatfs = fs.promises.statfs;
+const roomy = async () => ({ bsize: 4096, blocks: 25e6, bfree: 10e6, bavail: 10e6 });
+fs.promises.statfs = roomy;
+
+(async () => {
+ const rec = path.join(tmp, 'vod-live-1-1.mp4');
+ fs.writeFileSync(rec, Buffer.alloc(4096));
+ const sidecar = tools.seekableSidecarPath(rec);
+ const tmpOut = rec + '.live-remux.tmp.mp4';
+
+ // 1) One pass at a time: a second call while ffmpeg runs starts nothing.
+ const first = tools.remuxForLiveSeeking(rec);
+ assert.ok(await until(() => spawned.length === 1), 'the first call starts one pass');
+ assert.strictEqual(await tools.remuxForLiveSeeking(rec), false, 'a call while the pass runs does nothing');
+ assert.strictEqual(spawned.length, 1, 'no second ffmpeg on the same recording');
+ assert.strictEqual(spawned[0].args[spawned[0].args.length - 1], tmpOut, 'the pass writes the temp file');
+ assert.ok(spawned[0].args.includes('+faststart'), 'an MP4 snapshot is written with faststart');
+
+ // 2) The pass is not cut short at 30 s: its budget scales with the file, and it ends when ffmpeg ends.
+ assert.strictEqual(tools.remuxBudgetMs(1e6), 180000, 'a small file gets the 3 min floor');
+ assert.strictEqual(tools.remuxBudgetMs(8e9), 8 * 45000 + 60000, 'an 8 GB file gets ~45 s per GB');
+ assert.strictEqual(tools.remuxBudgetMs(500e9), 45 * 60000, 'the ceiling is 45 min');
+ spawned[0].finish();
+ assert.strictEqual(await first, true, 'a finished pass replaces the snapshot');
+ assert.strictEqual(fs.readFileSync(sidecar, 'utf8'), 'snapshot');
+ assert.ok(!fs.existsSync(tmpOut), 'the temp file became the snapshot');
+
+ // 3) Paced: the next pass waits 9x as long as the last took, so an immediate call does nothing;
+ // once that window has passed (however long the pass above took here), a new pass starts.
+ assert.strictEqual(await tools.remuxForLiveSeeking(rec), false, 'a call inside the pacing window does nothing');
+ assert.strictEqual(spawned.length, 1);
+ let second = null;
+ for (let i = 0; i < 200 && !second; i++) {
+ await new Promise((res) => setTimeout(res, 50));
+ const call = tools.remuxForLiveSeeking(rec); // not due: resolves false; due: runs until the stand-in ends
+ const r = await Promise.race([call.then(() => 'returned'), until(() => spawned.length === 2).then((ok) => (ok ? 'spawned' : 'none'))]);
+ if (r === 'spawned') second = call;
+ }
+ assert.ok(second && spawned.length === 2, 'after the pacing window a new pass starts');
+
+ // 4) Finalization retires the recording: the running pass is killed, nothing is left behind.
+ fs.writeFileSync(tmpOut, 'partial');
+ await tools.cleanupSeekableFile(rec);
+ assert.strictEqual(spawned[1].killedWith, 'SIGKILL', 'the live pass is stopped at finalization');
+ assert.strictEqual(await second, false);
+ assert.ok(!fs.existsSync(sidecar), 'the snapshot is removed');
+ assert.ok(!fs.existsSync(tmpOut), 'the killed pass removed its temp file');
+
+ // 5) No room: a pass that cannot fit its copy next to the recording is skipped (and starts no ffmpeg).
+ fs.promises.statfs = async () => ({ bsize: 4096, blocks: 25e6, bfree: 1e5, bavail: 1e5 }); // ~100 GB disk, 0.4 GB free
+ const big = path.join(tmp, 'vod-live-2-2.webm');
+ fs.writeFileSync(big, Buffer.alloc(1024));
+ const before = spawned.length;
+ assert.strictEqual(await tools.remuxForLiveSeeking(big), false, 'no room: skipped');
+ assert.strictEqual(spawned.length, before, 'no ffmpeg without room for its copy');
+ const final = await tools.remuxForSeekingDetailed(big);
+ assert.deepStrictEqual([final.ok, final.error], [false, 'no room for the remux copy'], 'the final remux checks the disk too');
+ assert.strictEqual(spawned.length, before);
+ assert.strictEqual(await tools.hasRoomFor(tmp, 0), false, '0.4 GB free is under the 10 GB reserve');
+ fs.promises.statfs = roomy;
+ assert.strictEqual(await tools.hasRoomFor(tmp, 20e9), true, '41 GB free takes a 20 GB copy and keeps 10 GB');
+ assert.strictEqual(await tools.hasRoomFor(tmp, 35e9), false, 'but not a 35 GB one');
+ fs.promises.statfs = realStatfs;
+
+ // 6) removeFile: off the event loop, never throws.
+ const gone = path.join(tmp, 'gone.bin');
+ fs.writeFileSync(gone, 'x');
+ assert.strictEqual(await tools.removeFile(gone), true);
+ assert.ok(!fs.existsSync(gone));
+ assert.strictEqual(await tools.removeFile(gone), true, 'a missing file is already gone');
+ assert.strictEqual(await tools.removeFile(null), false);
+ const aDir = path.join(tmp, 'a-dir');
+ fs.mkdirSync(aDir);
+ assert.strictEqual(await tools.removeFile(aDir), false, 'a directory is not removed');
+ assert.ok(fs.existsSync(aDir));
+
+ // 7) diskUsage reads statfs, not df through a shell.
+ const vodStorage = require('../server/vod/vod-storage');
+ const d = vodStorage.diskUsage(tmp);
+ assert.ok(d.total > 0 && d.available > 0 && d.used >= 0 && d.usePct > 0 && d.usePct <= 100, `diskUsage: ${JSON.stringify(d)}`);
+
+ // 8) The VOD lifecycle files hold no synchronous unlink/rename of recordings, and no execSync.
+ const src = (f) => fs.readFileSync(path.join(__dirname, '..', 'server', 'vod', f), 'utf8');
+ for (const f of ['media-tools.js', 'finalize.js', 'vod-storage.js', 'health-scanner.js']) {
+ const body = src(f).split('\n').filter((l) => !/^\s*(\/\/|\*)/.test(l)).join('\n');
+ const hits = (body.match(/\b(renameSync|execSync)\b/g) || []);
+ assert.deepStrictEqual(hits, [], `${f}: ${hits.join(', ')}`);
+ }
+ // cleanupTempFile (browser chunks) and the SDP files are small: the only unlinkSync left in these files.
+ for (const f of ['finalize.js', 'vod-storage.js', 'health-scanner.js']) {
+ assert.ok(!/unlinkSync/.test(src(f)), `${f} frees files with removeFile / fs.promises`);
+ }
+
+ fs.rmSync(tmp, { recursive: true, force: true });
+ console.log('live-remux-pacing: ok');
+ process.exit(0);
+})().catch((err) => { console.error(err); process.exit(1); });