openvibers/openvibe.media Commits

Reliability: fix the file-descriptor leak in media streaming (80 GB of deleted VODs pinned), size-scaled remux timeout, clip auto-retry with hot-fetch, stall watchdog

OP OpenVibers committed 15 Sept 2026 · 3036841684 · 8 files, +174 −45

- public/routes.js + thumbnail-service: every createReadStream is destroyed when the response closes
  (players abort thousands of range requests per session; each leaked an fd and, once the VOD was
  offloaded and unlinked, its whole file stayed on disk)
- remuxForSeeking: timeout ~45 s/GB (3 min floor) instead of flat 60 s — multi-hour recordings were
  killed mid-remux and left without a seek index, so cloud cuts scanned the whole file
- clips: cut_error / cut_attempts / cut_next_at; clip-jobs.js sweeper retries failed cuts (4 attempts,
  backoff), pulling cloud-stored VODs back to local disk from attempt 2 when space allows;
  manual re-cut resets the ladder; processing rows orphaned by a restart are requeued
- clip-cutter: stall-based watchdog + 20 min budget for cloud sources; real error text kept
- fd watchdog logs when the process holds > 1500 descriptors

Co-Authored-By: Claude Fable 5.1 <[email protected]>
Claude-Session: https://claude.ai/code/session_01CMxJm1dB8rdsaMSgNjVa82

Changes

ddiff --git a/server/db/database.js b/server/db/database.js
iindex f6c2d7f..83a128c 100644
---- a/server/db/database.js
++++ b/server/db/database.js
@@@ -72,7 +72,7 @@ function normalizeThumbnailUrls() {
function migrateColumns() {
const wanted = {
vods: [['managed_stream_id', 'INTEGER']],
-- clips: [['channel_user_id', 'INTEGER']],
++ clips: [['channel_user_id', 'INTEGER'], ['cut_error', 'TEXT'], ['cut_attempts', 'INTEGER DEFAULT 0'], ['cut_next_at', 'DATETIME']],
};
for (const [table, cols] of Object.entries(wanted)) {
const existing = database.prepare(`PRAGMA table_info(${table})`).all().map(c => c.name);
ddiff --git a/server/index.js b/server/index.js
iindex ac86d27..79238eb 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -101,6 +101,14 @@ vodStorage.checkProviders()
.catch(err => console.warn('[Boot] Provider check failed:', err.message));
vodStorage.start(); // tiering sweep
healthJob.start(); // health scan + quarantine cleanup
++require('./vod/clip-jobs').start(); // failed clip re-cuts (auto-retry, hot-fetch)
++// Descriptor watchdog: a leak here once pinned 80 GB of deleted recordings to the disk.
++every(5 * 60 * 1000, () => {
++ try {
++ const n = fs.readdirSync('/proc/self/fd').length;
++ if (n > 1500) console.warn(`[Boot] ${n} open file descriptors — investigate a stream/handle leak (lsof -p ${process.pid} +L1)`);
++ } catch { /* not linux */ }
++});
every(60 * 1000, () => recorder.checkDisk()); // disk guardian
every(60 * 60 * 1000, () => thumbService.cleanupOldThumbnails()); // stale live thumbs
ddiff --git a/server/public/routes.js b/server/public/routes.js
iindex 078aa67..6a0c241 100644
---- a/server/public/routes.js
++++ b/server/public/routes.js
@@@ -49,6 +49,21 @@ function canAccessPrivate(record, req) {
return false;
}
++/**
++ * Pipe a file (or a byte range of it) to the response WITHOUT leaking the descriptor.
++ * `createReadStream(...).pipe(res)` alone never closes the file when the client goes away
++ * mid-transfer — and video players do that thousands of times per session (every seek
++ * aborts a range request). Deleted-but-open VOD files once held 80 GB of disk that way.
++ */
++function sendFileStream(res, filePath, opts) {
++ const stream = fs.createReadStream(filePath, opts);
++ const done = () => { try { stream.destroy(); } catch { /* */ } };
++ res.on('close', done); res.on('error', done); res.on('finish', done);
++ stream.on('error', () => { try { res.destroy(); } catch { /* */ } });
++ stream.pipe(res);
++ return stream;
++}
++
function streamFileWithRange(req, res, filePath, extraHeaders = {}) {
const stat = fs.statSync(filePath);
const range = req.headers.range;
@@@ -71,7 +86,7 @@ function streamFileWithRange(req, res, filePath, extraHeaders = {}) {
'Content-Type': contentType,
...extraHeaders,
});
-- fs.createReadStream(filePath, { start, end: Math.min(end, stat.size - 1) }).pipe(res);
++ sendFileStream(res, filePath, { start, end: Math.min(end, stat.size - 1) });
} else {
res.writeHead(200, {
'Content-Length': stat.size,
@@@ -79,7 +94,7 @@ function streamFileWithRange(req, res, filePath, extraHeaders = {}) {
'Accept-Ranges': 'bytes',
...extraHeaders,
});
-- fs.createReadStream(filePath).pipe(res);
++ sendFileStream(res, filePath);
}
}
@@@ -382,7 +397,7 @@ router.get('/f/:key', (req, res) => {
'Content-Type': contentType,
...headers,
});
-- fs.createReadStream(filePath, { start, end }).pipe(res);
++ sendFileStream(res, filePath, { start, end });
} else {
res.writeHead(200, {
'Content-Length': stat.size,
@@@ -390,7 +405,7 @@ router.get('/f/:key', (req, res) => {
'Accept-Ranges': 'bytes',
...headers,
});
-- fs.createReadStream(filePath).pipe(res);
++ sendFileStream(res, filePath);
}
} catch (err) {
console.error('[Public] /f error:', err.message);
ddiff --git a/server/thumbnails/thumbnail-service.js b/server/thumbnails/thumbnail-service.js
iindex 5224ac9..2a7017f 100644
---- a/server/thumbnails/thumbnail-service.js
++++ b/server/thumbnails/thumbnail-service.js
@@@ -300,7 +300,11 @@ function serveThumbnail(req, res) {
'Cache-Control': cacheControl,
'X-Robots-Tag': 'noindex',
});
-- fs.createReadStream(filePath).pipe(res);
++ const stream = fs.createReadStream(filePath);
++ const done = () => { try { stream.destroy(); } catch { /* */ } };
++ res.on('close', done); res.on('finish', done); res.on('error', done);
++ stream.on('error', () => { try { res.destroy(); } catch { /* */ } });
++ stream.pipe(res);
} catch (err) {
res.status(500).json({ error: 'Failed to serve thumbnail' });
}
ddiff --git a/server/vod/clip-cutter.js b/server/vod/clip-cutter.js
iindex 8549942..a0c7b5c 100644
---- a/server/vod/clip-cutter.js
++++ b/server/vod/clip-cutter.js
@@@ -101,10 +101,22 @@ async function cutClipFile({ source, startTime, duration }) {
// Budget generously: this is a background job with no deadline, and the old
// dur*2 + 40s budget was tight enough that a burst of concurrent encodes on a
// busy box blew through it — which is exactly how a batch of clips failed at once.
-- const budgetMs = Math.round(dur) * 8000 + (isUrl ? 90000 : 45000);
-- const to = setTimeout(() => { timedOut = true; try { ff.kill('SIGKILL'); } catch { /* */ } resolve(false); }, budgetMs);
-- ff.on('close', (code) => { clearTimeout(to); exitCode = code; resolve(code === 0); });
-- ff.on('error', (e) => { clearTimeout(to); errTail += `\nspawn error: ${e.message}`; resolve(false); });
++ // Remote sources get a much bigger budget (a cloud seek can mean walking a big file),
++ // and a STALL watchdog rather than a flat kill: as long as the output keeps growing
++ // the cut is alive; only a file that stops growing for STALL_MS gets killed.
++ const budgetMs = Math.round(dur) * 8000 + (isUrl ? 20 * 60000 : 45000);
++ const STALL_MS = isUrl ? 4 * 60000 : 90000;
++ let lastSize = -1, lastGrowth = Date.now();
++ const started = Date.now();
++ const watchdog = setInterval(() => {
++ let size = 0; try { size = fs.statSync(outPath).size; } catch { size = 0; }
++ if (size !== lastSize) { lastSize = size; lastGrowth = Date.now(); }
++ const stalled = Date.now() - lastGrowth > STALL_MS;
++ const overBudget = Date.now() - started > budgetMs;
++ if (stalled || overBudget) { timedOut = true; errTail += `\n[watchdog] ${stalled ? 'no output growth for ' + Math.round(STALL_MS / 1000) + 's' : 'over budget'}`; try { ff.kill('SIGKILL'); } catch { /* */ } }
++ }, 5000);
++ ff.on('close', (code) => { clearInterval(watchdog); exitCode = code; resolve(!timedOut && code === 0); });
++ ff.on('error', (e) => { clearInterval(watchdog); errTail += `\nspawn error: ${e.message}`; resolve(false); });
_stderrRef.get = () => errTail;
}).finally(() => { _active = Math.max(0, _active - 1); });
@@@ -112,9 +124,7 @@ async function cutClipFile({ source, startTime, duration }) {
try { fs.existsSync(outPath) && fs.unlinkSync(outPath); } catch { /* */ }
// Surface ffmpeg's own last words — the real reason, not a generic label.
const tail = String(_stderrRef.get() || '').split('\n').filter(Boolean).slice(-3).join(' | ').slice(0, 400);
-- const why = timedOut
-- ? `ffmpeg timed out after ${Math.round((Math.round(dur) * 8000 + (isUrl ? 90000 : 45000)) / 1000)}s`
-- : `ffmpeg exited ${exitCode}`;
++ const why = timedOut ? `ffmpeg gave up (${isUrl ? 'cloud source' : 'local source'})` : `ffmpeg exited ${exitCode}`;
return { ok: false, error: tail ? `${why}: ${tail}` : why };
}
ddiff --git a/server/vod/clip-jobs.js b/server/vod/clip-jobs.js
nnew file mode 100644
iindex 0000000..279246a
---- /dev/null
++++ b/server/vod/clip-jobs.js
@@@ -0,0 +1,104 @@
++/**
++ * OpenVibe.Media — clip cut reliability.
++ *
++ * recutClip(clipId, { reason }) the ONE code path that (re)cuts an existing clip row:
++ * resolves the source, cuts, updates the row, thumbnails,
++ * fires the webhook. Records cut_error / cut_attempts and
++ * schedules an automatic retry on failure.
++ * start() sweeper: every 3 min, failed clips that still have
++ * attempts left are re-cut. From the second attempt on, a
++ * VOD that lives in cloud storage is first pulled back to
++ * local disk (moveToHot) — cutting a multi-hour recording
++ * over HTTP is what used to time out — as long as there is
++ * comfortable free space for it.
++ */
++'use strict';
++const fs = require('node:fs');
++const path = require('node:path');
++const db = require('../db/database');
++const config = require('../config');
++const cutter = require('./clip-cutter');
++const vodStorage = require('./vod-storage');
++const { sendWebhook } = require('../webhooks');
++
++const MAX_ATTEMPTS = 4;
++const BACKOFF_MIN = [2, 10, 30, 90]; // minutes before attempt 2, 3, 4, …
++const SWEEP_MS = 3 * 60 * 1000;
++const HOT_FETCH_FREE_MULTIPLE = 3; // need free space ≥ 3× the VOD size to pull it back
++let _timer = null, _busy = false;
++
++function freeBytes() {
++ try { const st = fs.statfsSync(path.resolve(config.vod.path)); return Number(st.bavail) * Number(st.bsize); } catch { return 0; }
++}
++function _clipPublic(clip) { try { return require('./clips-routes').clipPublic(clip); } catch { return clip; } }
++
++/** (Re)cut one clip row. Returns { ok, error }. */
++async function recutClip(clipId, { reason = 'recut' } = {}) {
++ const clip = db.getClipById(clipId);
++ if (!clip) return { ok: false, error: 'Clip not found' };
++ if (!clip.vod_id) return { ok: false, error: 'Clip has no source VOD' };
++ const vod = db.get('SELECT * FROM vods WHERE id = ?', [clip.vod_id]);
++ if (!vod) { db.run("UPDATE clips SET status = 'failed', cut_error = ?, cut_next_at = NULL WHERE id = ?", ['Source VOD no longer exists', clipId]); return { ok: false, error: 'Source VOD no longer exists' }; }
++ const attempt = (Number(clip.cut_attempts) || 0) + 1;
++ db.run("UPDATE clips SET status = 'processing', cut_attempts = ?, cut_next_at = NULL WHERE id = ?", [attempt, clipId]);
++
++ // Attempt 2+: bring a cloud-stored VOD home first when the disk can take it.
++ let source = await vodStorage.resolveMediaSource(vod);
++ if (source && source.kind === 'url' && attempt >= 2) {
++ const need = (Number(vod.file_size) || 0) * HOT_FETCH_FREE_MULTIPLE;
++ if (need && freeBytes() > need) {
++ console.log(`[Clips] clip ${clipId}: pulling vod ${vod.id} (${(vod.file_size / 1048576).toFixed(0)} MB) back to local disk for a reliable cut`);
++ const r = await vodStorage.moveToHot(vod.id).catch(e => ({ ok: false, error: e.message }));
++ if (r && r.ok) source = await vodStorage.resolveMediaSource(db.get('SELECT * FROM vods WHERE id = ?', [vod.id]));
++ else console.warn(`[Clips] clip ${clipId}: hot fetch failed: ${r && r.error}`);
++ } else {
++ console.log(`[Clips] clip ${clipId}: not enough free disk to pull vod ${vod.id} home — cutting from the cloud copy`);
++ }
++ }
++ if (!source) return fail(clipId, clip, attempt, 'VOD media unavailable (not on disk and no cloud copy)');
++
++ const startTime = Number(clip.start_time) || 0;
++ const duration = Math.max(1, (Number(clip.end_time) || 0) - startTime);
++ const cut = await cutter.cutClipFile({ source: source.value, startTime, duration });
++ if (!cut.ok) return fail(clipId, clip, attempt, cut.error);
++ db.run("UPDATE clips SET file_path = ?, duration_seconds = ?, end_time = ?, status = 'ready', cut_error = NULL, cut_next_at = NULL WHERE id = ?",
++ [cut.filePath, cut.duration, startTime + cut.duration, clipId]);
++ try { await require('../thumbnails/thumbnail-service').generateClipThumbnail(clipId, cut.filePath); } catch { /* */ }
++ console.log(`[Clips] Clip ${clipId} ${reason} OK from vod ${clip.vod_id} (${startTime.toFixed(1)}-${(startTime + cut.duration).toFixed(1)}s, attempt ${attempt})`);
++ sendWebhook(clip.app_id, 'clip.ready', _clipPublic(db.getClipById(clipId))).catch(() => {});
++ return { ok: true };
++}
++
++function fail(clipId, clip, attempt, error) {
++ const msg = String(error || 'cut failed').slice(0, 500);
++ const more = attempt < MAX_ATTEMPTS;
++ const mins = BACKOFF_MIN[Math.min(attempt - 1, BACKOFF_MIN.length - 1)];
++ const nextAt = more ? new Date(Date.now() + mins * 60000).toISOString().replace('T', ' ').slice(0, 19) : null;
++ db.run("UPDATE clips SET status = 'failed', cut_error = ?, cut_next_at = ? WHERE id = ?", [msg, nextAt, clipId]);
++ console.warn(`[Clips] Clip ${clipId} attempt ${attempt}/${MAX_ATTEMPTS} failed: ${msg}${more ? ` — retry in ${mins} min` : ' — giving up'}`);
++ sendWebhook(clip.app_id, 'clip.failed', _clipPublic(db.getClipById(clipId))).catch(() => {});
++ return { ok: false, error: msg, retry_at: nextAt };
++}
++
++async function sweep() {
++ if (_busy) return;
++ _busy = true;
++ try {
++ const due = db.all(`SELECT id FROM clips WHERE status = 'failed' AND vod_id IS NOT NULL
++ AND COALESCE(cut_attempts, 0) < ? AND (cut_next_at IS NULL OR cut_next_at <= CURRENT_TIMESTAMP)
++ AND created_at >= datetime('now', '-30 days') ORDER BY created_at DESC LIMIT 4`, [MAX_ATTEMPTS]) || [];
++ for (const row of due) { try { await recutClip(row.id, { reason: 'auto-retry' }); } catch (e) { console.warn(`[Clips] auto-retry ${row.id}:`, e.message); } }
++ } finally { _busy = false; }
++}
++
++function start() {
++ if (_timer) return;
++ // Anything left 'processing' by a crash/restart is really failed — queue it for a retry.
++ try { db.run("UPDATE clips SET status = 'failed', cut_error = COALESCE(cut_error, 'interrupted by a restart') WHERE status = 'processing'"); } catch { /* */ }
++ setTimeout(() => sweep().catch(() => {}), 40 * 1000);
++ _timer = setInterval(() => sweep().catch(e => console.warn('[Clips] retry sweep:', e.message)), SWEEP_MS);
++ if (_timer.unref) _timer.unref();
++ console.log('[Clips] cut retry sweeper started (every 3 min, up to 4 attempts, hot-fetch from attempt 2)');
++}
++
++module.exports = { recutClip, sweep, start, MAX_ATTEMPTS };
ddiff --git a/server/vod/clips-routes.js b/server/vod/clips-routes.js
iindex 728a74a..c4a0dda 100644
---- a/server/vod/clips-routes.js
++++ b/server/vod/clips-routes.js
@@@ -67,6 +67,10 @@ function clipPublic(clip) {
title: clip.title,
description: clip.description,
status: clip.status || (clip.file_path ? 'ready' : 'processing'),
++ cut_error: clip.cut_error || null,
++ cut_attempts: Number(clip.cut_attempts) || 0,
++ cut_next_at: clip.cut_next_at || null,
++ will_retry: clip.status === 'failed' && !!clip.cut_next_at,
start_time: clip.start_time,
end_time: clip.end_time,
duration: clip.duration_seconds || 0,
@@@ -199,8 +203,11 @@ router.post('/', tenantAuth({ allowUser: true }), clipUpload.single('video'), as
console.log(`[Clips] Clip ${clipId} cut from vod ${vodId} (${startTime.toFixed(1)}-${(startTime + cut.duration).toFixed(1)}s)`);
sendWebhook(appId, 'clip.ready', clipPublic(db.getClipById(clipId))).catch(() => {});
} else {
-- console.warn(`[Clips] Clip ${clipId} failed: ${cut.error}`);
-- db.run('UPDATE clips SET status = ? WHERE id = ?', ['failed', clipId]);
++ // First attempt failed: record why and let the retry sweeper take it from here
++ // (attempt 2 pulls a cloud-stored VOD back to local disk first).
++ const nextAt = new Date(Date.now() + 2 * 60000).toISOString().replace('T', ' ').slice(0, 19);
++ db.run("UPDATE clips SET status = 'failed', cut_error = ?, cut_attempts = 1, cut_next_at = ? WHERE id = ?", [String(cut.error || 'cut failed').slice(0, 500), nextAt, clipId]);
++ console.warn(`[Clips] Clip ${clipId} failed: ${cut.error} — auto-retry in 2 min`);
sendWebhook(appId, 'clip.failed', clipPublic(db.getClipById(clipId))).catch(() => {});
}
})().catch(err => console.error('[Clips] Background cut error:', err.message));
@@@ -325,34 +332,9 @@ router.post('/:id/recut', tenantAuth(), async (req, res) => {
if (!clip || clip.app_id !== req.appId) return res.status(404).json({ error: 'Clip not found' });
if (clip.status === 'processing') return res.status(409).json({ error: 'Clip is already being cut' });
if (!clip.vod_id) return res.status(422).json({ error: 'Clip has no source VOD to re-cut from' });
--
-- const vod = db.get('SELECT * FROM vods WHERE id = ?', [clip.vod_id]);
-- if (!vod) return res.status(404).json({ error: 'Source VOD no longer exists' });
--
-- const vodStorage = require('./vod-storage');
-- const source = await vodStorage.resolveMediaSource(vod);
-- if (!source) return res.status(404).json({ error: 'VOD media unavailable (not on disk and no cloud copy)' });
--
-- const startTime = Number(clip.start_time) || 0;
-- const duration = Math.max(1, (Number(clip.end_time) || 0) - startTime);
-- db.run('UPDATE clips SET status = ? WHERE id = ?', ['processing', clipId]);
-- const appId = req.appId;
--
-- (async () => {
-- const cut = await cutter.cutClipFile({ source: source.value, startTime, duration });
-- if (cut.ok) {
-- db.run('UPDATE clips SET file_path = ?, duration_seconds = ?, end_time = ?, status = ? WHERE id = ?',
-- [cut.filePath, cut.duration, startTime + cut.duration, 'ready', clipId]);
-- try { await require('../thumbnails/thumbnail-service').generateClipThumbnail(clipId, cut.filePath); } catch { /* */ }
-- console.log(`[Clips] Clip ${clipId} RE-CUT from vod ${clip.vod_id} (${startTime.toFixed(1)}-${(startTime + cut.duration).toFixed(1)}s)`);
-- sendWebhook(appId, 'clip.ready', clipPublic(db.getClipById(clipId))).catch(() => {});
-- } else {
-- console.warn(`[Clips] Clip ${clipId} re-cut failed: ${cut.error}`);
-- db.run('UPDATE clips SET status = ? WHERE id = ?', ['failed', clipId]);
-- sendWebhook(appId, 'clip.failed', clipPublic(db.getClipById(clipId))).catch(() => {});
-- }
-- })().catch(err => console.error('[Clips] Background re-cut error:', err.message));
--
++ // A manual retry resets the attempt counter so it gets the full ladder again.
++ db.run("UPDATE clips SET cut_attempts = 0 WHERE id = ?", [clipId]);
++ require('./clip-jobs').recutClip(clipId, { reason: 're-cut' }).catch(err => console.error('[Clips] Background re-cut error:', err.message));
res.status(202).json({ id: clipId, status: 'processing' });
} catch (err) {
console.error('[Clips] Re-cut error:', err.message);
ddiff --git a/server/vod/media-tools.js b/server/vod/media-tools.js
iindex 5208ddc..6b87237 100644
---- a/server/vod/media-tools.js
++++ b/server/vod/media-tools.js
@@@ -56,8 +56,14 @@ function remuxForSeeking(filePath) {
proc.on('error', () => resolve(false));
-- // Timeout: 60s should be plenty for copy-mode remux
-- setTimeout(() => { try { proc.kill(); } catch {} }, 60000);
++ // 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));
});
}