openvibers/openvibe.media Commits
VOD durations are measured, never estimated: finalize stores what the file says and vods.duration_source says where it came from. probe = ffprobe's container duration of the finished file; remux = the last packet time of a stream-copy pass (the finalize remux reports it, or a separate pass into the null muxer when the remux gave none), used when the header has no duration or claims far more than the packets (1.5x + 30 s: the header is corrupt); unknown = nothing measurable, stored as 0 with health_status needs_review (probe_failed or inflated_duration) and hidden, where the old code kept the wall clock. The recorder's own start time only bounds a value (longer than 1.5x its run + 30 s is refused); an orphan finalized after a restart has no bound and is never measured as now - created_at (it used to be, so a boot-time finalize of a 2-day-old row could store 2 days). The short, corrupt and stat-failure outcomes now also replace the recorder's live elapsed-time estimate (they left it in duration_seconds), as does the recorder's failed-finalize quarantine. A later finalize that measures a VOD lifts the quarantine a failed attempt left and restores the owner's visibility. Merge/remux failures are recorded as health issues (merge_failed, remux_failed) instead of being ignored. The health job's duration repair and the GET /vods/:id self-heal record source probe. test/vod-duration.test.js covers the choice, every outcome and a real ffmpeg orphan.
Changes
@@@ -161,6 +161,17 @@ MEDIA_APP_KEYS="live:key1,games:key2"`status`: `pending → recording → ready | failed` (derived; failures come fromhealth quarantine: corrupt / zero-byte / missing file).++**Durations are measured, never estimated.** Finalize stores what the file says and++records where it came from in `duration_source`: `probe` (ffprobe's container++duration), `remux` (the last packet time of a stream-copy pass, used when the header++has none or is far longer than the packets) or `unknown` (nothing measurable: the++duration is stored as **0**, `health_status` is `needs_review` with `probe_failed` or++`inflated_duration`, and the VOD is hidden until a later finalize measures it). The++wall clock only bounds a recording the recorder saw start (a value longer than 1.5x++its run + 30 s is refused); an orphan finalized after a restart has no bound and is++never measured as now - created_at. While recording, `duration` is the live elapsed++time; finalize replaces it.++**Recording formats** (inherited codec-passthrough behavior — VODs are *not*always `.webm`): RTMP and RTP-H.264 record by lossless stream copy into afragmented **`.mp4`** (audio → AAC on the RTP path); RTP-VP8/VP9 copies into@@@ -59,7 +59,7 @@ function normalizeThumbnailUrls() {// (CREATE TABLE IF NOT EXISTS won't add columns to an existing table).function migrateColumns() {const wanted = {-- vods: [['managed_stream_id', 'INTEGER'], ['object_id', 'TEXT']],++ vods: [['managed_stream_id', 'INTEGER'], ['object_id', 'TEXT'], ['duration_source', 'TEXT']],clips: [['channel_user_id', 'INTEGER'], ['cut_error', 'TEXT'], ['cut_attempts', 'INTEGER DEFAULT 0'], ['cut_next_at', 'DATETIME'], ['object_id', 'TEXT']],files: [['object_id', 'TEXT']],pastes: [['object_id', 'TEXT']],@@@ -451,10 +451,11 @@ function updateVodHealth(vodId, { status, score, issues = [], probeDuration, proreturn r;}--function repairVodDuration(vodId, duration, fileSize) {++/** A measured duration (source probe | remux) replaces the stored one. */++function repairVodDuration(vodId, duration, fileSize, source = 'probe') {return run(-- `UPDATE vods SET duration_seconds = ?, file_size = ?, probe_duration_seconds = ?, last_health_scan_at = datetime('now') WHERE id = ?`,-- [duration, fileSize, duration, vodId]++ `UPDATE vods SET duration_seconds = ?, file_size = ?, probe_duration_seconds = ?, duration_source = ?, last_health_scan_at = datetime('now') WHERE id = ?`,++ [duration, fileSize, duration, source, vodId]);}@@@ -34,6 +34,7 @@ CREATE TABLE IF NOT EXISTS vods (file_size INTEGER DEFAULT 0,duration_seconds INTEGER DEFAULT 0,probe_duration_seconds REAL DEFAULT 0,++ duration_source TEXT, -- probe | remux | unknown: where duration_seconds came from (NULL: before it was recorded)probe_format_json TEXT DEFAULT '',health_status TEXT DEFAULT 'unknown',health_score INTEGER DEFAULT 0,@@@ -50,6 +50,8 @@ function vodPublic(vod) {status: db.vodStatus(vod),duration: vod.duration_seconds || 0,duration_seconds: vod.duration_seconds || 0,++ // Where the duration came from: probe (ffprobe), remux (packet timeline) or unknown (stored 0).++ duration_source: vod.duration_source || null,file_size: vod.file_size || 0,// Basename only (server paths stay private) — the inherited SPA derives// its /file/<name> playback URL from this.@@@ -150,11 +152,27 @@ async function finalizeVod(vodId, opts = {}) {}}++// Health issues finalize itself records when it cannot settle a recording. A later successful++// finalize (the vod.finalize retry job) lifts the quarantine they caused.++const FINALIZE_ISSUES = ['probe_failed', 'inflated_duration', 'stat_failed', 'finalize_failed'];++++function _issuesOf(vod) {++ try { const v = JSON.parse(vod.health_issues_json || '[]'); return Array.isArray(v) ? v : []; } catch { return []; }++}++++/** Quarantined by finalize (not by the health job or an operator): a good finalize lifts it. */++function _finalizeQuarantined(vod) {++ return !!vod.quarantined_at && _issuesOf(vod).some(i => FINALIZE_ISSUES.includes(i));++}++async function _doFinalize(vodId, opts) {const vod = db.getVodById(vodId);if (!vod) return null;const filePath = vod.file_path;-- const startTime = opts.startTimeMs || new Date(String(vod.created_at).replace(' ', 'T') + 'Z').getTime();++ // The recorder's own start time bounds the duration (a container claiming more than the++ // recorder ran is corrupt). An orphan has none: created_at is when the row was made, and++ // now - created_at says nothing about the footage, so it is never used.++ const wallClockSeconds = opts.startTimeMs ? Math.max(0, Math.round((Date.now() - opts.startTimeMs) / 1000)) : null;// Ephemeral clips-only recording: it existed only to serve live clips. Never// published — delete the file (+ any offloaded object) and its row.@@@ -177,17 +195,17 @@ async function _doFinalize(vodId, opts) {return null;}++ const issues = [];// Merge any pending browser-chunk segments before remuxing.if (opts.segmentPath && opts.segmentPath !== filePath) {-- await tools.concatWebmFiles(filePath, opts.segmentPath);++ if (!(await tools.concatWebmFiles(filePath, opts.segmentPath))) issues.push('merge_failed');}-- await tools.mergePendingSegments(filePath);++ if (await tools.mergePendingSegments(filePath)) { if (!issues.includes('merge_failed')) issues.push('merge_failed'); }// A recording that never received media (e.g. an RTP ingest that was// started and stopped without packets) leaves a zero-byte file. There is// nothing to review, so treat it like a missing file: report the failure and-- // delete the row and file — the wall-clock fallback below must never mark it-- // ready, and a 0:00 ghost must never reach listings.++ // delete the row and file — a 0:00 ghost must never reach listings.if (tools.getFileSizeSafe(filePath) === 0) {console.warn(`[VOD] vod ${vodId}: zero-byte recording — deleting empty recording`);try { fs.unlinkSync(filePath); } catch { /* */ }@@@ -198,32 +216,26 @@ async function _doFinalize(vodId, opts) {return null;}-- // Remux for proper seeking support (fast copy-mode, no re-encode)-- await tools.remuxForSeeking(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);-- // Probe actual duration with ffprobe-- const wallClockSeconds = Math.max(0, Math.round((Date.now() - startTime) / 1000));-- let durationSeconds = wallClockSeconds;-- let probeFormatJson = JSON.stringify({});-- try {-- const probeInfo = await tools.probeVodInfo(filePath);-- if (probeInfo.duration > 0) {-- // Guard against a corrupt/inflated container duration: real footage can-- // never be meaningfully longer than the wall-clock recording time.-- if (wallClockSeconds > 0 && probeInfo.duration > wallClockSeconds * 1.5 + 30) {-- console.warn(`[VOD] vod ${vodId}: probed duration ${probeInfo.duration}s >> wall-clock ${wallClockSeconds}s — using wall-clock (inflated/corrupt container)`);-- durationSeconds = wallClockSeconds;-- } else {-- durationSeconds = probeInfo.duration;-- }-- }-- probeFormatJson = JSON.stringify(probeInfo.format || {});-- } catch (probeErr) {-- console.warn(`[VOD] ffprobe failed for vod ${vodId}:`, probeErr.message);-- }++ // 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) => {++ const probe = await tools.probeDuration(p);++ let packets = remuxS || 0;++ if (!packets && !(probe.seconds > 0)) packets = (await tools.streamCopyDuration(p)).seconds;++ return { probe, choice: tools.chooseDuration({ probeS: probe.seconds, remuxS: packets, wallS: wallClockSeconds }) };++ };++ let { probe, choice } = await measure(filePath, remux.ok ? remux.seconds : 0);++ let durationSeconds = choice.seconds;++ let probeFormatJson = JSON.stringify(probe.format || {});++ if (!probe.ok) console.warn(`[VOD] ffprobe failed for vod ${vodId}: ${probe.error}`);// ── Recover from the lossless master if the served webm came out truncated ──// Only WebM recordings have a separate lossless master; an RTMP/H.264 MP4 is@@@ -233,16 +245,19 @@ async function _doFinalize(vodId, opts) {: null;let masterDur = 0;if (masterPath && fs.existsSync(masterPath)) {-- try { const mi = await tools.probeVodInfo(masterPath); masterDur = mi.duration || 0; } catch { /* */ }++ try { const mi = await tools.probeDuration(masterPath); masterDur = mi.seconds || 0; } catch { /* */ }if (masterDur > 0 && masterDur > durationSeconds + 15 && masterDur > durationSeconds * 1.15) {console.warn(`[VOD] vod ${vodId}: webm ${durationSeconds}s is short vs master ${masterDur}s — rebuilding webm from master`);const ok = await rebuildWebmFromMaster(masterPath, filePath);if (ok) {-- try { await tools.remuxForSeeking(filePath); tools.cleanupSeekableFile(filePath); } catch { /* */ }-- try {-- const mi2 = await tools.probeVodInfo(filePath);-- if (mi2.duration > durationSeconds) { durationSeconds = mi2.duration; probeFormatJson = JSON.stringify(mi2.format || {}); }-- } catch { /* */ }++ let again = { ok: false, seconds: 0 };++ try { again = await tools.remuxForSeekingDetailed(filePath); tools.cleanupSeekableFile(filePath); } catch { /* */ }++ const rebuilt = await measure(filePath, again.ok ? again.seconds : 0);++ if (rebuilt.choice.seconds > durationSeconds) {++ ({ probe, choice } = rebuilt);++ durationSeconds = choice.seconds;++ probeFormatJson = JSON.stringify(probe.format || {});++ }console.log(`[VOD] vod ${vodId}: recovered from master → ${durationSeconds}s`);} else {console.warn(`[VOD] vod ${vodId}: master recovery failed — keeping master for manual recovery`);@@@ -250,10 +265,24 @@ async function _doFinalize(vodId, opts) {}}++ for (const i of choice.issues) if (!issues.includes(i)) issues.push(i);++ const durationSource = choice.source;++ const stored = Math.round(durationSeconds);++ const measured = durationSeconds > 0 ? durationSeconds : 0;++if (opts.ffmpegCorrupted) {console.warn(`[VOD] Finalized VOD ${vodId} marked corrupt by FFmpeg diagnostics; quarantining without deletion`);-- _commitVodOutcome(vod, 'vod.failed', () => db.run(`UPDATE vods SET is_recording = 0, health_status = ?, health_issues_json = ?, probe_duration_seconds = ?, probe_format_json = ?, last_health_scan_at = datetime('now'), quarantined_at = datetime('now'), is_public = 0 WHERE id = ?`,-- ['corrupt', JSON.stringify(['ffmpeg-corruption-detected']), durationSeconds > 0 ? durationSeconds : 0, probeFormatJson, vodId]));++ _commitVodOutcome(vod, 'vod.failed', () => db.run(`UPDATE vods SET is_recording = 0, duration_seconds = ?, duration_source = ?, health_status = ?, health_issues_json = ?, probe_duration_seconds = ?, probe_format_json = ?, last_health_scan_at = datetime('now'), quarantined_at = datetime('now'), is_public = 0 WHERE id = ?`,++ [stored, durationSource, 'corrupt', JSON.stringify(['ffmpeg-corruption-detected', ...issues]), measured, probeFormatJson, vodId]));++ return db.getVodById(vodId);++ }++++ // Nothing measurable (ffprobe and the packet pass both failed, or every value is impossible):++ // the stored duration is 0 and the recording waits for review, hidden. Never the wall clock.++ if (durationSource === 'unknown') {++ console.warn(`[VOD] vod ${vodId}: no measurable duration (${issues.join(', ')}) — stored 0, needs_review`);++ _commitVodOutcome(vod, 'vod.failed', () => db.run(`UPDATE vods SET is_recording = 0, duration_seconds = 0, duration_source = 'unknown', health_status = 'needs_review', health_issues_json = ?, probe_duration_seconds = 0, probe_format_json = ?, last_health_scan_at = datetime('now'), quarantined_at = datetime('now'), is_public = 0 WHERE id = ?`,++ [JSON.stringify(issues), probeFormatJson, vodId]));return db.getVodById(vodId);}@@@ -261,8 +290,8 @@ async function _doFinalize(vodId, opts) {const MIN_VOD_SECONDS = parseInt(process.env.MIN_VOD_SECONDS || '2', 10);if (durationSeconds < MIN_VOD_SECONDS) {console.log(`[VOD] Quarantining short vod ${vodId}: duration ${durationSeconds}s`);-- _commitVodOutcome(vod, 'vod.failed', () => db.run(`UPDATE vods SET is_recording = 0, health_status = ?, health_issues_json = ?, probe_duration_seconds = ?, probe_format_json = ?, last_health_scan_at = datetime('now'), quarantined_at = datetime('now'), is_public = 0 WHERE id = ?`,-- ['needs_review', JSON.stringify(['short_duration']), durationSeconds > 0 ? durationSeconds : 0, probeFormatJson, vodId]));++ _commitVodOutcome(vod, 'vod.failed', () => db.run(`UPDATE vods SET is_recording = 0, duration_seconds = ?, duration_source = ?, health_status = ?, health_issues_json = ?, probe_duration_seconds = ?, probe_format_json = ?, last_health_scan_at = datetime('now'), quarantined_at = datetime('now'), is_public = 0 WHERE id = ?`,++ [stored, durationSource, 'needs_review', JSON.stringify(['short_duration', ...issues]), measured, probeFormatJson, vodId]));return db.getVodById(vodId);}@@@ -270,8 +299,11 @@ async function _doFinalize(vodId, opts) {try {stat = fs.statSync(filePath);} catch (err) {++ // The file vanished between the probe and here. Record it (no wall-clock duration left++ // behind by the live updates) and leave it for review.console.error(`[VOD] Failed to stat finalized VOD ${vodId}:`, err.message);-- db.run('UPDATE vods SET is_recording = 0 WHERE id = ?', [vodId]);++ db.run(`UPDATE vods SET is_recording = 0, duration_seconds = 0, duration_source = 'unknown', health_status = 'needs_review',++ health_issues_json = ?, last_health_scan_at = datetime('now') WHERE id = ?`, [JSON.stringify(['stat_failed', ...issues]), vodId]);return null;}// The lossless .master.mkv archive is only a fallback. Delete it ONLY once@@@ -304,15 +336,19 @@ async function _doFinalize(vodId, opts) {console.warn(`[VOD] Thumbnail generation failed for vod ${vodId}:`, err.message);}++ // A quarantine an earlier finalize attempt left (nothing measurable then) is lifted: the++ // recording is back to the visibility its owner chose.++ const lift = _finalizeQuarantined(db.getVodById(vodId) || vod);++// The ready transition and the vod.ready event commit together (then the webhook goes out).// A failed commit throws, as the plain UPDATE did before.announce(vod.app_id, 'vod.ready', {-- change: () => db.run('UPDATE vods SET is_recording = 0, duration_seconds = ?, file_size = ?, probe_duration_seconds = ?, probe_format_json = ?, health_status = ? WHERE id = ?',-- [durationSeconds, stat.size, durationSeconds, probeFormatJson, 'ok', vodId]),++ change: () => db.run(`UPDATE vods SET is_recording = 0, duration_seconds = ?, duration_source = ?, file_size = ?, probe_duration_seconds = ?, probe_format_json = ?, health_status = ?, health_issues_json = ?${lift ? ", quarantined_at = NULL, is_public = CASE WHEN COALESCE(visibility, 'public') = 'public' THEN 1 ELSE 0 END" : ''} WHERE id = ?`,++ [stored, durationSource, stat.size, measured, probeFormatJson, 'ok', JSON.stringify(issues), vodId]),payload: () => vodPublic(db.getVodById(vodId)),});-- console.log(`[VOD] Finalized: vod ${vodId}, ${durationSeconds}s, ${(stat.size / 1024 / 1024).toFixed(1)}MB`);++ console.log(`[VOD] Finalized: vod ${vodId}, ${stored}s (${durationSource}), ${(stat.size / 1024 / 1024).toFixed(1)}MB`);return db.getVodById(vodId);}--module.exports = { finalizeVod, isFinalizing, vodPublic, rebuildWebmFromMaster, _absUrl };++module.exports = { finalizeVod, isFinalizing, vodPublic, rebuildWebmFromMaster, _absUrl, FINALIZE_ISSUES };@@@ -11,50 +11,76 @@ const path = require('path');const fs = require('fs');const { spawn } = require('child_process');++/**++ * Seconds of the last `time=HH:MM:SS.xx` progress stamp in ffmpeg's stderr, or 0. A stream-copy++ * run reports the timestamp of the last packet it wrote, which is the media's real length even++ * when the container header has no (or a wrong) duration.++ */++function lastProgressSeconds(text) {++ const re = /time=\s*(-?\d+):(\d{2}):(\d{2}(?:\.\d+)?)/g;++ let m, last = null;++ while ((m = re.exec(String(text || '')))) last = m;++ if (!last) return 0;++ const s = Number(last[1]) * 3600 + Number(last[2]) * 60 + Number(last[3]);++ return Number.isFinite(s) && s > 0 ? s : 0;++}++++/** Keep the tail of a growing ffmpeg stderr (a multi-hour run prints megabytes of progress). */++function tailAppend(buf, chunk, max = 16384) {++ const out = buf + chunk;++ return out.length > max ? out.slice(-max) : out;++}++/*** 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.++ *++ * 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 remuxForSeeking(filePath) {++function remuxForSeekingDetailed(filePath) {return new Promise((resolve) => {const ext = path.extname(filePath).toLowerCase();-- if (ext !== '.webm' && ext !== '.mp4') return resolve(false); // WebM + MP4 only++ if (ext !== '.webm' && ext !== '.mp4') return resolve({ ok: false, seconds: 0, error: 'unsupported container' }); // WebM + MP4 onlyconst tmpPath = filePath + '.remux' + ext;const outArgs = ext === '.mp4'? ['-c', 'copy', '-movflags', '+faststart', '-fflags', '+genpts']: ['-c', 'copy', '-fflags', '+genpts'];-- const proc = spawn('ffmpeg', [-- '-y', '-i', filePath,-- ...outArgs,-- tmpPath,-- ], { stdio: ['ignore', 'ignore', 'pipe'] });++ 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 += d);++ 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(true);++ resolve({ ok: true, seconds: lastProgressSeconds(stderr), error: null });} catch (err) {console.warn(`[VOD] Remux rename failed:`, err.message);try { fs.unlinkSync(tmpPath); } catch {}-- resolve(false);++ 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(false);++ resolve({ ok: false, seconds: 0, error: `ffmpeg remux exited ${code}` });}});-- proc.on('error', () => resolve(false));++ proc.on('error', (err) => resolve({ ok: false, seconds: 0, error: err.message }));// 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@@@ -67,6 +93,89 @@ function remuxForSeeking(filePath) {});}++/** remuxForSeekingDetailed(), answering true/false (the callers that only need the outcome). */++function remuxForSeeking(filePath) {++ return remuxForSeekingDetailed(filePath).then(r => r.ok);++}++++const isUrl = (src) => /^https?:\/\//i.test(String(src || ''));++// A remote source (a presigned B2/R2 URL) is read with ranged HTTP requests; nothing else is allowed.++const urlArgs = (src) => (isUrl(src) ? ['-protocol_whitelist', 'https,http,tls,tcp', '-rw_timeout', '30000000'] : []);++++/**++ * ffprobe a local file or a presigned https URL (ranged reads). Resolves++ * { ok, seconds (float, 0 when the container states no duration), format, streams, error };++ * ok is false when ffprobe could not read the source at all.++ */++function probeDuration(src, { timeoutMs = 10000 } = {}) {++ return new Promise((resolve) => {++ let proc, out = '', done = false;++ const finish = (r) => { if (done) return; done = true; clearTimeout(killer); resolve(r); };++ try {++ proc = spawn('ffprobe', ['-v', 'quiet', ...urlArgs(src), '-print_format', 'json', '-show_format', '-show_streams', String(src)],++ { stdio: ['ignore', 'pipe', 'ignore'] });++ } catch (err) { return resolve({ ok: false, seconds: 0, format: null, streams: [], error: err.message }); }++ const killer = setTimeout(() => { try { proc.kill('SIGKILL'); } catch {} finish({ ok: false, seconds: 0, format: null, streams: [], error: `ffprobe timed out after ${Math.round(timeoutMs / 1000)} s` }); }, timeoutMs);++ proc.stdout.on('data', d => { out += d; });++ proc.on('error', (err) => finish({ ok: false, seconds: 0, format: null, streams: [], error: err.message }));++ proc.on('close', (code) => {++ let info = null;++ try { info = JSON.parse(out); } catch { /* unreadable */ }++ if (!info || !info.format) return finish({ ok: false, seconds: 0, format: null, streams: [], error: `ffprobe could not read the source (exit ${code})` });++ const d = parseFloat(info.format.duration);++ finish({ ok: true, seconds: Number.isFinite(d) && d > 0 ? d : 0, format: info.format, streams: info.streams || [], error: null });++ });++ });++}++++/**++ * The media's length measured by reading every packet: a stream-copy pass into the null muxer++ * (nothing is written). Resolves { ok, seconds, error }. For containers whose header has no++ * duration (a live WebM, a fragmented MP4 that was never remuxed) and to cross-check one that++ * looks wrong. Reads the whole source: a remote one is downloaded in full.++ */++function streamCopyDuration(src, { timeoutMs = 20 * 60 * 1000 } = {}) {++ return new Promise((resolve) => {++ let proc, err = '', done = false;++ const finish = (r) => { if (done) return; done = true; clearTimeout(killer); resolve(r); };++ try {++ proc = spawn('ffmpeg', ['-nostdin', '-hide_banner', ...urlArgs(src), '-i', String(src), '-map', '0', '-c', 'copy', '-f', 'null', '-'],++ { stdio: ['ignore', 'ignore', 'pipe'] });++ } catch (e) { return resolve({ ok: false, seconds: 0, error: e.message }); }++ const killer = setTimeout(() => { try { proc.kill('SIGKILL'); } catch {} finish({ ok: false, seconds: 0, error: `stream-copy pass timed out after ${Math.round(timeoutMs / 1000)} s` }); }, timeoutMs);++ proc.stderr.on('data', d => { err = tailAppend(err, String(d)); });++ proc.on('error', (e) => finish({ ok: false, seconds: 0, error: e.message }));++ proc.on('close', (code) => {++ const seconds = code === 0 ? lastProgressSeconds(err) : 0;++ finish(seconds > 0 ? { ok: true, seconds, error: null } : { ok: false, seconds: 0, error: `stream-copy pass gave no duration (exit ${code})` });++ });++ });++}++++/**++ * The stored duration of a recording, from what was measured — never from the wall clock.++ * probeS ffprobe's container duration of the finished file (0 = none / unreadable)++ * remuxS the last packet time of a stream-copy pass over it (0 = none)++ * wallS how long the recorder ran, when it is known (a bound only, never a value)++ * Returns { seconds, source: 'probe' | 'remux' | 'unknown', issues }. A container header far++ * longer than the packet timeline is corrupt, so the packets win; a value longer than the++ * recorder ran (1.5x + 30 s) is impossible, so it is refused. Nothing usable: 0, 'unknown'.++ */++function chooseDuration({ probeS = 0, remuxS = 0, wallS = null } = {}) {++ const issues = [];++ let probe = probeS > 0 ? probeS : 0;++ const remux = remuxS > 0 ? remuxS : 0;++ if (probe && remux && probe > remux * 1.5 + 30) { issues.push('inflated_container_duration'); probe = 0; }++ const limit = wallS > 0 ? wallS * 1.5 + 30 : Infinity;++ for (const [s, source] of [[probe, 'probe'], [remux, 'remux']]) {++ if (!s) continue;++ if (s > limit) { if (!issues.includes('inflated_duration')) issues.push('inflated_duration'); continue; }++ return { seconds: s, source, issues };++ }++ if (!probeS && !remuxS) issues.push('probe_failed');++ return { seconds: 0, source: 'unknown', issues };++}++/** Probe a media file's start_time (seconds), or 0 if probing fails. */function probeStartTime(filePath) {return new Promise((resolve) => {@@@ -295,17 +404,26 @@ function concatWebmFiles(basePath, appendPath) {})();}++/** Merge pending browser-chunk segments; resolves the number that could not be merged. */async function mergePendingSegments(filePath) {++ let failed = 0;for (const segmentPath of getPendingSegmentFiles(filePath)) {const ok = await concatWebmFiles(filePath, segmentPath);if (!ok) {++ failed++;console.warn(`[VOD] Failed to merge segment into ${path.basename(filePath)}: ${path.basename(segmentPath)}`);}}++ return failed;}module.exports = {remuxForSeeking,++ remuxForSeekingDetailed,++ lastProgressSeconds,++ probeDuration,++ streamCopyDuration,++ chooseDuration,remuxForLiveSeeking,seekableSidecarPath,cleanupSeekableFile,@@@ -261,7 +261,8 @@ class StreamRecorder {require('./finalize').finalizeVod(vodId, { startTimeMs: rec?.startTime }).catch(() => {// Never bare-mark a failed recording as ready — that published// 0:00 ghosts. Quarantine it out of listings instead.-- db.run(`UPDATE vods SET is_recording = 0, health_status = 'needs_review',++ // The live duration updates were wall-clock estimates: none of it is kept.++ db.run(`UPDATE vods SET is_recording = 0, duration_seconds = 0, duration_source = 'unknown', health_status = 'needs_review',health_issues_json = ?, quarantined_at = datetime('now'), is_public = 0 WHERE id = ?`,[JSON.stringify(['finalize_failed']), vodId]);});@@@ -327,7 +327,7 @@ router.get('/:id', tenantAuth({ allowUser: true }), async (req, res) => {const duration = await tools.probeVodDuration(vod.file_path);if (duration > 0) {const fileSize = tools.getFileSizeSafe(vod.file_path);-- db.run('UPDATE vods SET duration_seconds = ?, file_size = ? WHERE id = ?', [duration, fileSize, vod.id]);++ db.run("UPDATE vods SET duration_seconds = ?, file_size = ?, duration_source = 'probe' WHERE id = ?", [duration, fileSize, vod.id]);vod.duration_seconds = duration;vod.file_size = fileSize;}nnew file mode 100644@@@ -0,0 +1,184 @@++'use strict';++// VOD duration truthfulness (server/vod/finalize.js, media-tools.chooseDuration): the stored duration++// is what the file says (ffprobe, else the packet timeline of a stream-copy pass), never the wall++// clock; vods.duration_source says which (probe | remux | unknown). Nothing measurable stores 0 with++// needs_review; an orphan (no recorder start time) is never measured as now - created_at; a later++// good finalize lifts the quarantine a failed one left.++const assert = require('assert');++const fs = require('fs');++const os = require('os');++const path = require('path');++const { spawnSync } = require('child_process');++++const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-duration-'));++const dir = (n) => { const d = path.join(tmp, n); fs.mkdirSync(d, { recursive: true }); return d; };++process.env.DB_PATH = path.join(tmp, 'media.db');++process.env.VOD_PATH = dir('vods');++process.env.CLIPS_PATH = dir('clips');++process.env.THUMBNAILS_PATH = dir('thumbnails');++process.env.FILES_PATH = dir('files');++process.env.PASTES_PATH = dir('pastes');++process.env.OBJECTS_PATH = dir('objects');++process.env.MEDIA_JOBS_ENABLED = '0';++++const tools = require('../server/vod/media-tools');++++// ── 1. chooseDuration: measured values only, bounded by the recorder's run when known ──++{++ const c = tools.chooseDuration;++ assert.deepStrictEqual(c({ probeS: 600.4 }), { seconds: 600.4, source: 'probe', issues: [] });++ assert.deepStrictEqual(c({ probeS: 0, remuxS: 590 }), { seconds: 590, source: 'remux', issues: [] }, 'no container duration: the packets');++ assert.deepStrictEqual(c({}), { seconds: 0, source: 'unknown', issues: ['probe_failed'] }, 'nothing measured: 0, unknown');++ const inflated = c({ probeS: 86400, remuxS: 600 });++ assert.deepStrictEqual([inflated.seconds, inflated.source], [600, 'remux'], 'a header far longer than the packets is corrupt');++ assert.ok(inflated.issues.includes('inflated_container_duration'));++ const bounded = c({ probeS: 5000, remuxS: 0, wallS: 600 });++ assert.deepStrictEqual([bounded.seconds, bounded.source], [0, 'unknown'], 'longer than the recorder ran: refused, and the wall clock is NOT used');++ assert.ok(bounded.issues.includes('inflated_duration'));++ assert.strictEqual(c({ probeS: 5000, remuxS: 590, wallS: 600 }).source, 'remux', 'the plausible measurement wins');++ assert.strictEqual(c({ probeS: 172800 }).seconds, 172800, 'no start time (orphan): no bound either, only measurements');++ assert.strictEqual(tools.lastProgressSeconds('frame=1 time=00:00:01.00 x\rframe=2 time=01:02:03.50 bitrate'), 3723.5);++ assert.strictEqual(tools.lastProgressSeconds('time=N/A'), 0);++ console.log('✅ chooseDuration: probe, else the packet timeline, else 0/unknown; bounded by the recorder run; never the wall clock');++}++++const db = require('../server/db/database');++const finalize = require('../server/vod/finalize');++db.upsertApp({ app_id: 'live', api_key: 'live-key' });++++const cols = db.all('PRAGMA table_info(vods)').map(c => c.name);++assert.ok(cols.includes('duration_source'), 'vods.duration_source exists');++++const daysAgo = (d) => new Date(Date.now() - d * 864e5).toISOString().replace('T', ' ').slice(0, 19);++function makeVod(file, { createdAt = daysAgo(2), wallDuration = 99999 } = {}) {++ const r = db.run(`INSERT INTO vods (app_id, user_id, title, file_path, is_recording, duration_seconds, created_at, is_public, visibility)++ VALUES ('live', 7, 'rec', ?, 1, ?, ?, 1, 'public')`, [file, wallDuration, createdAt]);++ return Number(r.lastInsertRowid);++}++const row = (id) => db.get('SELECT * FROM vods WHERE id = ?', [id]);++++const realTools = { ...tools };++function restore() { Object.assign(tools, realTools); }++++(async () => {++ // ── 2. Probe failure: 0 + needs_review + hidden; the live wall-clock estimate is not kept ──++ {++ const file = path.join(process.env.VOD_PATH, 'garbage.mp4');++ fs.writeFileSync(file, Buffer.alloc(20000, 7));++ const id = makeVod(file);++ tools.remuxForSeekingDetailed = async () => ({ ok: false, seconds: 0, error: 'ffmpeg remux exited 1' });++ tools.probeDuration = async () => ({ ok: false, seconds: 0, format: null, streams: [], error: 'unreadable' });++ tools.streamCopyDuration = async () => ({ ok: false, seconds: 0, error: 'no packets' });++ await finalize.finalizeVod(id);++ const v = row(id);++ assert.strictEqual(v.duration_seconds, 0, 'probe failure stores 0, not the recorder\'s wall-clock estimate');++ assert.strictEqual(v.duration_source, 'unknown');++ assert.strictEqual(v.health_status, 'needs_review');++ const issues = JSON.parse(v.health_issues_json);++ assert.ok(issues.includes('probe_failed') && issues.includes('remux_failed'), `issues: ${issues}`);++ assert.strictEqual(v.is_recording, 0);++ assert.strictEqual(v.is_public, 0, 'hidden until reviewed');++ assert.ok(v.quarantined_at);++++ // A later finalize that can measure it lifts the quarantine (the vod.finalize retry job does this).++ tools.probeDuration = async () => ({ ok: true, seconds: 1234.4, format: { format_name: 'mp4' }, streams: [], error: null });++ await finalize.finalizeVod(id);++ const w = row(id);++ assert.deepStrictEqual([w.duration_seconds, w.duration_source, w.health_status], [1234, 'probe', 'ok']);++ assert.strictEqual(w.quarantined_at, null, 'quarantine lifted');++ assert.strictEqual(w.is_public, 1, 'back to the visibility its owner chose');++ restore();++ console.log('✅ probe failure: duration 0, duration_source unknown, needs_review, hidden; a later good finalize lifts it');++ }++++ // ── 3. Orphans (no start time) never use now - created_at; a bound applies only with a real start ──++ {++ const file = path.join(process.env.VOD_PATH, 'orphan.mp4');++ fs.writeFileSync(file, Buffer.alloc(20000, 1));++ const id = makeVod(file, { createdAt: daysAgo(3) });++ tools.remuxForSeekingDetailed = async () => ({ ok: true, seconds: 3601.2, error: null });++ tools.probeDuration = async () => ({ ok: true, seconds: 3600.8, format: {}, streams: [], error: null });++ await finalize.finalizeVod(id);++ const v = row(id);++ assert.deepStrictEqual([v.duration_seconds, v.duration_source], [3601, 'probe'], 'orphan: the measured length, not 3 days');++++ const file2 = path.join(process.env.VOD_PATH, 'inflated.mp4');++ fs.writeFileSync(file2, Buffer.alloc(20000, 2));++ const id2 = makeVod(file2);++ tools.probeDuration = async () => ({ ok: true, seconds: 50000, format: {}, streams: [], error: null });++ tools.remuxForSeekingDetailed = async () => ({ ok: true, seconds: 50000, error: null });++ await finalize.finalizeVod(id2, { startTimeMs: Date.now() - 600 * 1000 });++ const w = row(id2);++ assert.deepStrictEqual([w.duration_seconds, w.duration_source, w.health_status], [0, 'unknown', 'needs_review'],++ 'longer than the recorder ran: 0 and review, where the old code stored the wall clock');++ assert.ok(JSON.parse(w.health_issues_json).includes('inflated_duration'));++++ const file3 = path.join(process.env.VOD_PATH, 'header.mp4');++ fs.writeFileSync(file3, Buffer.alloc(20000, 3));++ const id3 = makeVod(file3);++ tools.probeDuration = async () => ({ ok: true, seconds: 90000, format: {}, streams: [], error: null });++ tools.remuxForSeekingDetailed = async () => ({ ok: true, seconds: 581, error: null });++ await finalize.finalizeVod(id3, { startTimeMs: Date.now() - 600 * 1000 });++ assert.deepStrictEqual([row(id3).duration_seconds, row(id3).duration_source, row(id3).health_status], [581, 'remux', 'ok'], 'the packet timeline when the header lies');++ restore();++ console.log('✅ orphans are measured, never now - created_at; impossible values need review; a lying header yields to the packets');++ }++++ // ── 4. Short, corrupt and stat-failure paths also drop the wall-clock estimate ──++ {++ const file = path.join(process.env.VOD_PATH, 'short.mp4');++ fs.writeFileSync(file, Buffer.alloc(20000, 4));++ const id = makeVod(file);++ tools.remuxForSeekingDetailed = async () => ({ ok: true, seconds: 1, error: null });++ tools.probeDuration = async () => ({ ok: true, seconds: 1, format: {}, streams: [], error: null });++ await finalize.finalizeVod(id);++ assert.deepStrictEqual([row(id).duration_seconds, row(id).duration_source, row(id).health_status], [1, 'probe', 'needs_review']);++++ const file2 = path.join(process.env.VOD_PATH, 'corrupt.mp4');++ fs.writeFileSync(file2, Buffer.alloc(20000, 5));++ const id2 = makeVod(file2);++ tools.probeDuration = async () => ({ ok: true, seconds: 42, format: {}, streams: [], error: null });++ tools.remuxForSeekingDetailed = async () => ({ ok: true, seconds: 42, error: null });++ await finalize.finalizeVod(id2, { ffmpegCorrupted: true });++ assert.deepStrictEqual([row(id2).duration_seconds, row(id2).duration_source, row(id2).health_status], [42, 'probe', 'corrupt'],++ 'the corrupt path stores the measurement, not the live estimate');++++ const file3 = path.join(process.env.VOD_PATH, 'vanishing.mp4');++ fs.writeFileSync(file3, Buffer.alloc(20000, 6));++ const id3 = makeVod(file3);++ tools.probeDuration = async (p) => { if (p === file3) { try { fs.unlinkSync(file3); } catch { /* */ } } return { ok: true, seconds: 300, format: {}, streams: [], error: null }; };++ tools.remuxForSeekingDetailed = async () => ({ ok: true, seconds: 300, error: null });++ await finalize.finalizeVod(id3);++ const v3 = row(id3);++ assert.deepStrictEqual([v3.is_recording, v3.duration_seconds, v3.duration_source, v3.health_status], [0, 0, 'unknown', 'needs_review']);++ assert.ok(JSON.parse(v3.health_issues_json).includes('stat_failed'));++ restore();++ console.log('✅ short, corrupt and stat-failure outcomes never keep the wall-clock estimate');++ }++++ // ── 5. Real ffmpeg: an orphaned 6 s recording made 2 days ago finalizes to ~6 s (probe) ──++ if (spawnSync('ffmpeg', ['-version']).status !== 0) {++ console.log('⚠️ ffmpeg not found: the real-file finalize check is skipped');++ } else {++ const file = path.join(process.env.VOD_PATH, 'real.mp4');++ const mk = spawnSync('ffmpeg', ['-v', 'error', '-y', '-f', 'lavfi', '-i', 'testsrc=duration=6:size=160x120:rate=10',++ '-f', 'lavfi', '-i', 'sine=frequency=440:duration=6', '-c:v', 'libx264', '-pix_fmt', 'yuv420p', '-c:a', 'aac', '-shortest',++ '-movflags', '+frag_keyframe+empty_moov+default_base_moof', '-f', 'mp4', file]);++ assert.strictEqual(mk.status, 0, String(mk.stderr));++ const id = makeVod(file, { createdAt: daysAgo(2) });++ await finalize.finalizeVod(id);++ const v = row(id);++ assert.ok(v.duration_seconds >= 5 && v.duration_seconds <= 7, `measured ${v.duration_seconds}s`);++ assert.strictEqual(v.duration_source, 'probe');++ assert.strictEqual(v.health_status, 'ok');++ const packets = await tools.streamCopyDuration(file);++ assert.ok(packets.ok && packets.seconds > 5 && packets.seconds < 7, `stream-copy pass measured ${packets.seconds}`);++ const pr = await tools.probeDuration(path.join(tmp, 'nope.mp4'));++ assert.strictEqual(pr.ok, false, 'an unreadable source is a failed probe, not a zero-length one');++ console.log('✅ a real orphaned recording finalizes to its measured length (probe), and the packet pass agrees');++ }++++ db.close();++ fs.rmSync(tmp, { recursive: true, force: true });++ console.log('\n✅ All VOD duration tests passed');++})().catch((err) => { console.error(err); process.exit(1); });
@@ -161,6 +161,17 @@ MEDIA_APP_KEYS="live:key1,games:key2"
@@ -161,6 +161,17 @@ MEDIA_APP_KEYS="live:key1,games:key2"
`status`: `pending → recording → ready | failed` (derived; failures come from
`status`: `pending → recording → ready | failed` (derived; failures come from
health quarantine: corrupt / zero-byte / missing file).
health quarantine: corrupt / zero-byte / missing file).
+**Durations are measured, never estimated.** Finalize stores what the file says and
+records where it came from in `duration_source`: `probe` (ffprobe's container
+duration), `remux` (the last packet time of a stream-copy pass, used when the header
+has none or is far longer than the packets) or `unknown` (nothing measurable: the
+duration is stored as **0**, `health_status` is `needs_review` with `probe_failed` or
+`inflated_duration`, and the VOD is hidden until a later finalize measures it). The
+wall clock only bounds a recording the recorder saw start (a value longer than 1.5x
+its run + 30 s is refused); an orphan finalized after a restart has no bound and is
+never measured as now - created_at. While recording, `duration` is the live elapsed
+time; finalize replaces it.
+
**Recording formats** (inherited codec-passthrough behavior — VODs are *not*
**Recording formats** (inherited codec-passthrough behavior — VODs are *not*
always `.webm`): RTMP and RTP-H.264 record by lossless stream copy into a
always `.webm`): RTMP and RTP-H.264 record by lossless stream copy into a
fragmented **`.mp4`** (audio → AAC on the RTP path); RTP-VP8/VP9 copies into
fragmented **`.mp4`** (audio → AAC on the RTP path); RTP-VP8/VP9 copies into
@@ -59,7 +59,7 @@ function normalizeThumbnailUrls() {
@@ -59,7 +59,7 @@ function normalizeThumbnailUrls() {
// (CREATE TABLE IF NOT EXISTS won't add columns to an existing table).
// (CREATE TABLE IF NOT EXISTS won't add columns to an existing table).
function migrateColumns() {
function migrateColumns() {
const wanted = {
const wanted = {
- vods: [['managed_stream_id', 'INTEGER'], ['object_id', 'TEXT']],
+ vods: [['managed_stream_id', 'INTEGER'], ['object_id', 'TEXT'], ['duration_source', 'TEXT']],
clips: [['channel_user_id', 'INTEGER'], ['cut_error', 'TEXT'], ['cut_attempts', 'INTEGER DEFAULT 0'], ['cut_next_at', 'DATETIME'], ['object_id', 'TEXT']],
clips: [['channel_user_id', 'INTEGER'], ['cut_error', 'TEXT'], ['cut_attempts', 'INTEGER DEFAULT 0'], ['cut_next_at', 'DATETIME'], ['object_id', 'TEXT']],
files: [['object_id', 'TEXT']],
files: [['object_id', 'TEXT']],
pastes: [['object_id', 'TEXT']],
pastes: [['object_id', 'TEXT']],
@@ -451,10 +451,11 @@ function updateVodHealth(vodId, { status, score, issues = [], probeDuration, pro
@@ -451,10 +451,11 @@ function updateVodHealth(vodId, { status, score, issues = [], probeDuration, pro
return r;
return r;
}
}
-function repairVodDuration(vodId, duration, fileSize) {
+/** A measured duration (source probe | remux) replaces the stored one. */
+function repairVodDuration(vodId, duration, fileSize, source = 'probe') {
return run(
return run(
- `UPDATE vods SET duration_seconds = ?, file_size = ?, probe_duration_seconds = ?, last_health_scan_at = datetime('now') WHERE id = ?`,
- [duration, fileSize, duration, vodId]
+ `UPDATE vods SET duration_seconds = ?, file_size = ?, probe_duration_seconds = ?, duration_source = ?, last_health_scan_at = datetime('now') WHERE id = ?`,
+ [duration, fileSize, duration, source, vodId]
);
);
}
}
@@ -34,6 +34,7 @@ CREATE TABLE IF NOT EXISTS vods (
@@ -34,6 +34,7 @@ CREATE TABLE IF NOT EXISTS vods (
file_size INTEGER DEFAULT 0,
file_size INTEGER DEFAULT 0,
duration_seconds INTEGER DEFAULT 0,
duration_seconds INTEGER DEFAULT 0,
probe_duration_seconds REAL DEFAULT 0,
probe_duration_seconds REAL DEFAULT 0,
+ duration_source TEXT, -- probe | remux | unknown: where duration_seconds came from (NULL: before it was recorded)
probe_format_json TEXT DEFAULT '',
probe_format_json TEXT DEFAULT '',
health_status TEXT DEFAULT 'unknown',
health_status TEXT DEFAULT 'unknown',
health_score INTEGER DEFAULT 0,
health_score INTEGER DEFAULT 0,
@@ -50,6 +50,8 @@ function vodPublic(vod) {
@@ -50,6 +50,8 @@ function vodPublic(vod) {
status: db.vodStatus(vod),
status: db.vodStatus(vod),
duration: vod.duration_seconds || 0,
duration: vod.duration_seconds || 0,
duration_seconds: vod.duration_seconds || 0,
duration_seconds: vod.duration_seconds || 0,
+ // Where the duration came from: probe (ffprobe), remux (packet timeline) or unknown (stored 0).
+ duration_source: vod.duration_source || null,
file_size: vod.file_size || 0,
file_size: vod.file_size || 0,
// Basename only (server paths stay private) — the inherited SPA derives
// Basename only (server paths stay private) — the inherited SPA derives
// its /file/<name> playback URL from this.
// its /file/<name> playback URL from this.
@@ -150,11 +152,27 @@ async function finalizeVod(vodId, opts = {}) {
@@ -150,11 +152,27 @@ async function finalizeVod(vodId, opts = {}) {
}
}
}
}
+// Health issues finalize itself records when it cannot settle a recording. A later successful
+// finalize (the vod.finalize retry job) lifts the quarantine they caused.
+const FINALIZE_ISSUES = ['probe_failed', 'inflated_duration', 'stat_failed', 'finalize_failed'];
+
+function _issuesOf(vod) {
+ try { const v = JSON.parse(vod.health_issues_json || '[]'); return Array.isArray(v) ? v : []; } catch { return []; }
+}
+
+/** Quarantined by finalize (not by the health job or an operator): a good finalize lifts it. */
+function _finalizeQuarantined(vod) {
+ return !!vod.quarantined_at && _issuesOf(vod).some(i => FINALIZE_ISSUES.includes(i));
+}
+
async function _doFinalize(vodId, opts) {
async function _doFinalize(vodId, opts) {
const vod = db.getVodById(vodId);
const vod = db.getVodById(vodId);
if (!vod) return null;
if (!vod) return null;
const filePath = vod.file_path;
const filePath = vod.file_path;
- const startTime = opts.startTimeMs || new Date(String(vod.created_at).replace(' ', 'T') + 'Z').getTime();
+ // The recorder's own start time bounds the duration (a container claiming more than the
+ // recorder ran is corrupt). An orphan has none: created_at is when the row was made, and
+ // now - created_at says nothing about the footage, so it is never used.
+ const wallClockSeconds = opts.startTimeMs ? Math.max(0, Math.round((Date.now() - opts.startTimeMs) / 1000)) : null;
// 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.
@@ -177,17 +195,17 @@ async function _doFinalize(vodId, opts) {
@@ -177,17 +195,17 @@ async function _doFinalize(vodId, opts) {
return null;
return null;
}
}
+ const issues = [];
// Merge any pending browser-chunk segments before remuxing.
// Merge any pending browser-chunk segments before remuxing.
if (opts.segmentPath && opts.segmentPath !== filePath) {
if (opts.segmentPath && opts.segmentPath !== filePath) {
- await tools.concatWebmFiles(filePath, opts.segmentPath);
+ if (!(await tools.concatWebmFiles(filePath, opts.segmentPath))) issues.push('merge_failed');
}
}
- await tools.mergePendingSegments(filePath);
+ if (await tools.mergePendingSegments(filePath)) { if (!issues.includes('merge_failed')) issues.push('merge_failed'); }
// A recording that never received media (e.g. an RTP ingest that was
// A recording that never received media (e.g. an RTP ingest that was
// started and stopped without packets) leaves a zero-byte file. There is
// started and stopped without packets) leaves a zero-byte file. There is
// nothing to review, so treat it like a missing file: report the failure and
// nothing to review, so treat it like a missing file: report the failure and
- // delete the row and file — the wall-clock fallback below must never mark it
- // ready, and a 0:00 ghost must never reach listings.
+ // delete the row and file — a 0:00 ghost must never reach listings.
if (tools.getFileSizeSafe(filePath) === 0) {
if (tools.getFileSizeSafe(filePath) === 0) {
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 { fs.unlinkSync(filePath); } catch { /* */ }
@@ -198,32 +216,26 @@ async function _doFinalize(vodId, opts) {
@@ -198,32 +216,26 @@ async function _doFinalize(vodId, opts) {
return null;
return null;
}
}
- // Remux for proper seeking support (fast copy-mode, no re-encode)
- await tools.remuxForSeeking(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)
// Clean up the live seekable copy (no longer needed after final remux)
tools.cleanupSeekableFile(filePath);
tools.cleanupSeekableFile(filePath);
- // Probe actual duration with ffprobe
- const wallClockSeconds = Math.max(0, Math.round((Date.now() - startTime) / 1000));
- let durationSeconds = wallClockSeconds;
- let probeFormatJson = JSON.stringify({});
- try {
- const probeInfo = await tools.probeVodInfo(filePath);
- if (probeInfo.duration > 0) {
- // Guard against a corrupt/inflated container duration: real footage can
- // never be meaningfully longer than the wall-clock recording time.
- if (wallClockSeconds > 0 && probeInfo.duration > wallClockSeconds * 1.5 + 30) {
- console.warn(`[VOD] vod ${vodId}: probed duration ${probeInfo.duration}s >> wall-clock ${wallClockSeconds}s — using wall-clock (inflated/corrupt container)`);
- durationSeconds = wallClockSeconds;
- } else {
- durationSeconds = probeInfo.duration;
- }
- }
- probeFormatJson = JSON.stringify(probeInfo.format || {});
- } catch (probeErr) {
- console.warn(`[VOD] ffprobe failed for vod ${vodId}:`, probeErr.message);
- }
+ // 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) => {
+ const probe = await tools.probeDuration(p);
+ let packets = remuxS || 0;
+ if (!packets && !(probe.seconds > 0)) packets = (await tools.streamCopyDuration(p)).seconds;
+ return { probe, choice: tools.chooseDuration({ probeS: probe.seconds, remuxS: packets, wallS: wallClockSeconds }) };
+ };
+ let { probe, choice } = await measure(filePath, remux.ok ? remux.seconds : 0);
+ let durationSeconds = choice.seconds;
+ let probeFormatJson = JSON.stringify(probe.format || {});
+ if (!probe.ok) console.warn(`[VOD] ffprobe failed for vod ${vodId}: ${probe.error}`);
// ── Recover from the lossless master if the served webm came out truncated ──
// ── Recover from the lossless master if the served webm came out truncated ──
// Only WebM recordings have a separate lossless master; an RTMP/H.264 MP4 is
// Only WebM recordings have a separate lossless master; an RTMP/H.264 MP4 is
@@ -233,16 +245,19 @@ async function _doFinalize(vodId, opts) {
@@ -233,16 +245,19 @@ async function _doFinalize(vodId, opts) {
: null;
: null;
let masterDur = 0;
let masterDur = 0;
if (masterPath && fs.existsSync(masterPath)) {
if (masterPath && fs.existsSync(masterPath)) {
- try { const mi = await tools.probeVodInfo(masterPath); masterDur = mi.duration || 0; } catch { /* */ }
+ try { const mi = await tools.probeDuration(masterPath); masterDur = mi.seconds || 0; } catch { /* */ }
if (masterDur > 0 && masterDur > durationSeconds + 15 && masterDur > durationSeconds * 1.15) {
if (masterDur > 0 && masterDur > durationSeconds + 15 && masterDur > durationSeconds * 1.15) {
console.warn(`[VOD] vod ${vodId}: webm ${durationSeconds}s is short vs master ${masterDur}s — rebuilding webm from master`);
console.warn(`[VOD] vod ${vodId}: webm ${durationSeconds}s is short vs master ${masterDur}s — rebuilding webm from master`);
const ok = await rebuildWebmFromMaster(masterPath, filePath);
const ok = await rebuildWebmFromMaster(masterPath, filePath);
if (ok) {
if (ok) {
- try { await tools.remuxForSeeking(filePath); tools.cleanupSeekableFile(filePath); } catch { /* */ }
- try {
- const mi2 = await tools.probeVodInfo(filePath);
- if (mi2.duration > durationSeconds) { durationSeconds = mi2.duration; probeFormatJson = JSON.stringify(mi2.format || {}); }
- } catch { /* */ }
+ let again = { ok: false, seconds: 0 };
+ try { again = await tools.remuxForSeekingDetailed(filePath); tools.cleanupSeekableFile(filePath); } catch { /* */ }
+ const rebuilt = await measure(filePath, again.ok ? again.seconds : 0);
+ if (rebuilt.choice.seconds > durationSeconds) {
+ ({ probe, choice } = rebuilt);
+ durationSeconds = choice.seconds;
+ probeFormatJson = JSON.stringify(probe.format || {});
+ }
console.log(`[VOD] vod ${vodId}: recovered from master → ${durationSeconds}s`);
console.log(`[VOD] vod ${vodId}: recovered from master → ${durationSeconds}s`);
} else {
} else {
console.warn(`[VOD] vod ${vodId}: master recovery failed — keeping master for manual recovery`);
console.warn(`[VOD] vod ${vodId}: master recovery failed — keeping master for manual recovery`);
@@ -250,10 +265,24 @@ async function _doFinalize(vodId, opts) {
@@ -250,10 +265,24 @@ async function _doFinalize(vodId, opts) {
}
}
}
}
+ for (const i of choice.issues) if (!issues.includes(i)) issues.push(i);
+ const durationSource = choice.source;
+ const stored = Math.round(durationSeconds);
+ const measured = durationSeconds > 0 ? durationSeconds : 0;
+
if (opts.ffmpegCorrupted) {
if (opts.ffmpegCorrupted) {
console.warn(`[VOD] Finalized VOD ${vodId} marked corrupt by FFmpeg diagnostics; quarantining without deletion`);
console.warn(`[VOD] Finalized VOD ${vodId} marked corrupt by FFmpeg diagnostics; quarantining without deletion`);
- _commitVodOutcome(vod, 'vod.failed', () => db.run(`UPDATE vods SET is_recording = 0, health_status = ?, health_issues_json = ?, probe_duration_seconds = ?, probe_format_json = ?, last_health_scan_at = datetime('now'), quarantined_at = datetime('now'), is_public = 0 WHERE id = ?`,
- ['corrupt', JSON.stringify(['ffmpeg-corruption-detected']), durationSeconds > 0 ? durationSeconds : 0, probeFormatJson, vodId]));
+ _commitVodOutcome(vod, 'vod.failed', () => db.run(`UPDATE vods SET is_recording = 0, duration_seconds = ?, duration_source = ?, health_status = ?, health_issues_json = ?, probe_duration_seconds = ?, probe_format_json = ?, last_health_scan_at = datetime('now'), quarantined_at = datetime('now'), is_public = 0 WHERE id = ?`,
+ [stored, durationSource, 'corrupt', JSON.stringify(['ffmpeg-corruption-detected', ...issues]), measured, probeFormatJson, vodId]));
+ return db.getVodById(vodId);
+ }
+
+ // Nothing measurable (ffprobe and the packet pass both failed, or every value is impossible):
+ // the stored duration is 0 and the recording waits for review, hidden. Never the wall clock.
+ if (durationSource === 'unknown') {
+ console.warn(`[VOD] vod ${vodId}: no measurable duration (${issues.join(', ')}) — stored 0, needs_review`);
+ _commitVodOutcome(vod, 'vod.failed', () => db.run(`UPDATE vods SET is_recording = 0, duration_seconds = 0, duration_source = 'unknown', health_status = 'needs_review', health_issues_json = ?, probe_duration_seconds = 0, probe_format_json = ?, last_health_scan_at = datetime('now'), quarantined_at = datetime('now'), is_public = 0 WHERE id = ?`,
+ [JSON.stringify(issues), probeFormatJson, vodId]));
return db.getVodById(vodId);
return db.getVodById(vodId);
}
}
@@ -261,8 +290,8 @@ async function _doFinalize(vodId, opts) {
@@ -261,8 +290,8 @@ async function _doFinalize(vodId, opts) {
const MIN_VOD_SECONDS = parseInt(process.env.MIN_VOD_SECONDS || '2', 10);
const MIN_VOD_SECONDS = parseInt(process.env.MIN_VOD_SECONDS || '2', 10);
if (durationSeconds < MIN_VOD_SECONDS) {
if (durationSeconds < MIN_VOD_SECONDS) {
console.log(`[VOD] Quarantining short vod ${vodId}: duration ${durationSeconds}s`);
console.log(`[VOD] Quarantining short vod ${vodId}: duration ${durationSeconds}s`);
- _commitVodOutcome(vod, 'vod.failed', () => db.run(`UPDATE vods SET is_recording = 0, health_status = ?, health_issues_json = ?, probe_duration_seconds = ?, probe_format_json = ?, last_health_scan_at = datetime('now'), quarantined_at = datetime('now'), is_public = 0 WHERE id = ?`,
- ['needs_review', JSON.stringify(['short_duration']), durationSeconds > 0 ? durationSeconds : 0, probeFormatJson, vodId]));
+ _commitVodOutcome(vod, 'vod.failed', () => db.run(`UPDATE vods SET is_recording = 0, duration_seconds = ?, duration_source = ?, health_status = ?, health_issues_json = ?, probe_duration_seconds = ?, probe_format_json = ?, last_health_scan_at = datetime('now'), quarantined_at = datetime('now'), is_public = 0 WHERE id = ?`,
+ [stored, durationSource, 'needs_review', JSON.stringify(['short_duration', ...issues]), measured, probeFormatJson, vodId]));
return db.getVodById(vodId);
return db.getVodById(vodId);
}
}
@@ -270,8 +299,11 @@ async function _doFinalize(vodId, opts) {
@@ -270,8 +299,11 @@ async function _doFinalize(vodId, opts) {
try {
try {
stat = fs.statSync(filePath);
stat = fs.statSync(filePath);
} catch (err) {
} catch (err) {
+ // The file vanished between the probe and here. Record it (no wall-clock duration left
+ // behind by the live updates) and leave it for review.
console.error(`[VOD] Failed to stat finalized VOD ${vodId}:`, err.message);
console.error(`[VOD] Failed to stat finalized VOD ${vodId}:`, err.message);
- db.run('UPDATE vods SET is_recording = 0 WHERE id = ?', [vodId]);
+ db.run(`UPDATE vods SET is_recording = 0, duration_seconds = 0, duration_source = 'unknown', health_status = 'needs_review',
+ health_issues_json = ?, last_health_scan_at = datetime('now') WHERE id = ?`, [JSON.stringify(['stat_failed', ...issues]), vodId]);
return null;
return null;
}
}
// The lossless .master.mkv archive is only a fallback. Delete it ONLY once
// The lossless .master.mkv archive is only a fallback. Delete it ONLY once
@@ -304,15 +336,19 @@ async function _doFinalize(vodId, opts) {
@@ -304,15 +336,19 @@ async function _doFinalize(vodId, opts) {
console.warn(`[VOD] Thumbnail generation failed for vod ${vodId}:`, err.message);
console.warn(`[VOD] Thumbnail generation failed for vod ${vodId}:`, err.message);
}
}
+ // A quarantine an earlier finalize attempt left (nothing measurable then) is lifted: the
+ // recording is back to the visibility its owner chose.
+ const lift = _finalizeQuarantined(db.getVodById(vodId) || vod);
+
// The ready transition and the vod.ready event commit together (then the webhook goes out).
// The ready transition and the vod.ready event commit together (then the webhook goes out).
// A failed commit throws, as the plain UPDATE did before.
// A failed commit throws, as the plain UPDATE did before.
announce(vod.app_id, 'vod.ready', {
announce(vod.app_id, 'vod.ready', {
- change: () => db.run('UPDATE vods SET is_recording = 0, duration_seconds = ?, file_size = ?, probe_duration_seconds = ?, probe_format_json = ?, health_status = ? WHERE id = ?',
- [durationSeconds, stat.size, durationSeconds, probeFormatJson, 'ok', vodId]),
+ change: () => db.run(`UPDATE vods SET is_recording = 0, duration_seconds = ?, duration_source = ?, file_size = ?, probe_duration_seconds = ?, probe_format_json = ?, health_status = ?, health_issues_json = ?${lift ? ", quarantined_at = NULL, is_public = CASE WHEN COALESCE(visibility, 'public') = 'public' THEN 1 ELSE 0 END" : ''} WHERE id = ?`,
+ [stored, durationSource, stat.size, measured, probeFormatJson, 'ok', JSON.stringify(issues), vodId]),
payload: () => vodPublic(db.getVodById(vodId)),
payload: () => vodPublic(db.getVodById(vodId)),
});
});
- console.log(`[VOD] Finalized: vod ${vodId}, ${durationSeconds}s, ${(stat.size / 1024 / 1024).toFixed(1)}MB`);
+ console.log(`[VOD] Finalized: vod ${vodId}, ${stored}s (${durationSource}), ${(stat.size / 1024 / 1024).toFixed(1)}MB`);
return db.getVodById(vodId);
return db.getVodById(vodId);
}
}
-module.exports = { finalizeVod, isFinalizing, vodPublic, rebuildWebmFromMaster, _absUrl };
+module.exports = { finalizeVod, isFinalizing, vodPublic, rebuildWebmFromMaster, _absUrl, FINALIZE_ISSUES };
@@ -11,50 +11,76 @@ const path = require('path');
@@ -11,50 +11,76 @@ const path = require('path');
const fs = require('fs');
const fs = require('fs');
const { spawn } = require('child_process');
const { spawn } = require('child_process');
+/**
+ * Seconds of the last `time=HH:MM:SS.xx` progress stamp in ffmpeg's stderr, or 0. A stream-copy
+ * run reports the timestamp of the last packet it wrote, which is the media's real length even
+ * when the container header has no (or a wrong) duration.
+ */
+function lastProgressSeconds(text) {
+ const re = /time=\s*(-?\d+):(\d{2}):(\d{2}(?:\.\d+)?)/g;
+ let m, last = null;
+ while ((m = re.exec(String(text || '')))) last = m;
+ if (!last) return 0;
+ const s = Number(last[1]) * 3600 + Number(last[2]) * 60 + Number(last[3]);
+ return Number.isFinite(s) && s > 0 ? s : 0;
+}
+
+/** Keep the tail of a growing ffmpeg stderr (a multi-hour run prints megabytes of progress). */
+function tailAppend(buf, chunk, max = 16384) {
+ const out = buf + chunk;
+ return out.length > max ? out.slice(-max) : out;
+}
+
/**
/**
* 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.
+ *
+ * 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 remuxForSeeking(filePath) {
+function remuxForSeekingDetailed(filePath) {
return new Promise((resolve) => {
return new Promise((resolve) => {
const ext = path.extname(filePath).toLowerCase();
const ext = path.extname(filePath).toLowerCase();
- if (ext !== '.webm' && ext !== '.mp4') return resolve(false); // WebM + MP4 only
+ if (ext !== '.webm' && ext !== '.mp4') return resolve({ ok: false, seconds: 0, error: 'unsupported container' }); // WebM + MP4 only
const tmpPath = filePath + '.remux' + ext;
const tmpPath = filePath + '.remux' + ext;
const outArgs = ext === '.mp4'
const outArgs = ext === '.mp4'
? ['-c', 'copy', '-movflags', '+faststart', '-fflags', '+genpts']
? ['-c', 'copy', '-movflags', '+faststart', '-fflags', '+genpts']
: ['-c', 'copy', '-fflags', '+genpts'];
: ['-c', 'copy', '-fflags', '+genpts'];
- const proc = spawn('ffmpeg', [
- '-y', '-i', filePath,
- ...outArgs,
- tmpPath,
- ], { stdio: ['ignore', 'ignore', 'pipe'] });
+ 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 = '';
let stderr = '';
- proc.stderr.on('data', d => stderr += d);
+ proc.stderr.on('data', d => { stderr = tailAppend(stderr, String(d)); });
proc.on('close', (code) => {
proc.on('close', (code) => {
if (code === 0 && fs.existsSync(tmpPath)) {
if (code === 0 && fs.existsSync(tmpPath)) {
try {
try {
fs.renameSync(tmpPath, filePath);
fs.renameSync(tmpPath, filePath);
console.log(`[VOD] Remuxed for seeking: ${path.basename(filePath)}`);
console.log(`[VOD] Remuxed for seeking: ${path.basename(filePath)}`);
- resolve(true);
+ resolve({ ok: true, seconds: lastProgressSeconds(stderr), error: null });
} catch (err) {
} catch (err) {
console.warn(`[VOD] Remux rename failed:`, err.message);
console.warn(`[VOD] Remux rename failed:`, err.message);
try { fs.unlinkSync(tmpPath); } catch {}
try { fs.unlinkSync(tmpPath); } catch {}
- resolve(false);
+ resolve({ ok: false, seconds: 0, error: err.message });
}
}
} else {
} else {
console.warn(`[VOD] Remux failed (code ${code}): ${stderr.slice(-200)}`);
console.warn(`[VOD] Remux failed (code ${code}): ${stderr.slice(-200)}`);
try { if (fs.existsSync(tmpPath)) fs.unlinkSync(tmpPath); } catch {}
try { if (fs.existsSync(tmpPath)) fs.unlinkSync(tmpPath); } catch {}
- resolve(false);
+ resolve({ ok: false, seconds: 0, error: `ffmpeg remux exited ${code}` });
}
}
});
});
- proc.on('error', () => resolve(false));
+ proc.on('error', (err) => resolve({ ok: false, seconds: 0, error: err.message }));
// Timeout scales with size: a copy-mode remux still rewrites every byte (twice for
// 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
// +faststart), and a flat 60 s killed every multi-hour recording — which left them
@@ -67,6 +93,89 @@ function remuxForSeeking(filePath) {
@@ -67,6 +93,89 @@ function remuxForSeeking(filePath) {
});
});
}
}
+/** remuxForSeekingDetailed(), answering true/false (the callers that only need the outcome). */
+function remuxForSeeking(filePath) {
+ return remuxForSeekingDetailed(filePath).then(r => r.ok);
+}
+
+const isUrl = (src) => /^https?:\/\//i.test(String(src || ''));
+// A remote source (a presigned B2/R2 URL) is read with ranged HTTP requests; nothing else is allowed.
+const urlArgs = (src) => (isUrl(src) ? ['-protocol_whitelist', 'https,http,tls,tcp', '-rw_timeout', '30000000'] : []);
+
+/**
+ * ffprobe a local file or a presigned https URL (ranged reads). Resolves
+ * { ok, seconds (float, 0 when the container states no duration), format, streams, error };
+ * ok is false when ffprobe could not read the source at all.
+ */
+function probeDuration(src, { timeoutMs = 10000 } = {}) {
+ return new Promise((resolve) => {
+ let proc, out = '', done = false;
+ const finish = (r) => { if (done) return; done = true; clearTimeout(killer); resolve(r); };
+ try {
+ proc = spawn('ffprobe', ['-v', 'quiet', ...urlArgs(src), '-print_format', 'json', '-show_format', '-show_streams', String(src)],
+ { stdio: ['ignore', 'pipe', 'ignore'] });
+ } catch (err) { return resolve({ ok: false, seconds: 0, format: null, streams: [], error: err.message }); }
+ const killer = setTimeout(() => { try { proc.kill('SIGKILL'); } catch {} finish({ ok: false, seconds: 0, format: null, streams: [], error: `ffprobe timed out after ${Math.round(timeoutMs / 1000)} s` }); }, timeoutMs);
+ proc.stdout.on('data', d => { out += d; });
+ proc.on('error', (err) => finish({ ok: false, seconds: 0, format: null, streams: [], error: err.message }));
+ proc.on('close', (code) => {
+ let info = null;
+ try { info = JSON.parse(out); } catch { /* unreadable */ }
+ if (!info || !info.format) return finish({ ok: false, seconds: 0, format: null, streams: [], error: `ffprobe could not read the source (exit ${code})` });
+ const d = parseFloat(info.format.duration);
+ finish({ ok: true, seconds: Number.isFinite(d) && d > 0 ? d : 0, format: info.format, streams: info.streams || [], error: null });
+ });
+ });
+}
+
+/**
+ * The media's length measured by reading every packet: a stream-copy pass into the null muxer
+ * (nothing is written). Resolves { ok, seconds, error }. For containers whose header has no
+ * duration (a live WebM, a fragmented MP4 that was never remuxed) and to cross-check one that
+ * looks wrong. Reads the whole source: a remote one is downloaded in full.
+ */
+function streamCopyDuration(src, { timeoutMs = 20 * 60 * 1000 } = {}) {
+ return new Promise((resolve) => {
+ let proc, err = '', done = false;
+ const finish = (r) => { if (done) return; done = true; clearTimeout(killer); resolve(r); };
+ try {
+ proc = spawn('ffmpeg', ['-nostdin', '-hide_banner', ...urlArgs(src), '-i', String(src), '-map', '0', '-c', 'copy', '-f', 'null', '-'],
+ { stdio: ['ignore', 'ignore', 'pipe'] });
+ } catch (e) { return resolve({ ok: false, seconds: 0, error: e.message }); }
+ const killer = setTimeout(() => { try { proc.kill('SIGKILL'); } catch {} finish({ ok: false, seconds: 0, error: `stream-copy pass timed out after ${Math.round(timeoutMs / 1000)} s` }); }, timeoutMs);
+ proc.stderr.on('data', d => { err = tailAppend(err, String(d)); });
+ proc.on('error', (e) => finish({ ok: false, seconds: 0, error: e.message }));
+ proc.on('close', (code) => {
+ const seconds = code === 0 ? lastProgressSeconds(err) : 0;
+ finish(seconds > 0 ? { ok: true, seconds, error: null } : { ok: false, seconds: 0, error: `stream-copy pass gave no duration (exit ${code})` });
+ });
+ });
+}
+
+/**
+ * The stored duration of a recording, from what was measured — never from the wall clock.
+ * probeS ffprobe's container duration of the finished file (0 = none / unreadable)
+ * remuxS the last packet time of a stream-copy pass over it (0 = none)
+ * wallS how long the recorder ran, when it is known (a bound only, never a value)
+ * Returns { seconds, source: 'probe' | 'remux' | 'unknown', issues }. A container header far
+ * longer than the packet timeline is corrupt, so the packets win; a value longer than the
+ * recorder ran (1.5x + 30 s) is impossible, so it is refused. Nothing usable: 0, 'unknown'.
+ */
+function chooseDuration({ probeS = 0, remuxS = 0, wallS = null } = {}) {
+ const issues = [];
+ let probe = probeS > 0 ? probeS : 0;
+ const remux = remuxS > 0 ? remuxS : 0;
+ if (probe && remux && probe > remux * 1.5 + 30) { issues.push('inflated_container_duration'); probe = 0; }
+ const limit = wallS > 0 ? wallS * 1.5 + 30 : Infinity;
+ for (const [s, source] of [[probe, 'probe'], [remux, 'remux']]) {
+ if (!s) continue;
+ if (s > limit) { if (!issues.includes('inflated_duration')) issues.push('inflated_duration'); continue; }
+ return { seconds: s, source, issues };
+ }
+ if (!probeS && !remuxS) issues.push('probe_failed');
+ return { seconds: 0, source: 'unknown', issues };
+}
+
/** Probe a media file's start_time (seconds), or 0 if probing fails. */
/** Probe a media file's start_time (seconds), or 0 if probing fails. */
function probeStartTime(filePath) {
function probeStartTime(filePath) {
return new Promise((resolve) => {
return new Promise((resolve) => {
@@ -295,17 +404,26 @@ function concatWebmFiles(basePath, appendPath) {
@@ -295,17 +404,26 @@ function concatWebmFiles(basePath, appendPath) {
})();
})();
}
}
+/** Merge pending browser-chunk segments; resolves the number that could not be merged. */
async function mergePendingSegments(filePath) {
async function mergePendingSegments(filePath) {
+ let failed = 0;
for (const segmentPath of getPendingSegmentFiles(filePath)) {
for (const segmentPath of getPendingSegmentFiles(filePath)) {
const ok = await concatWebmFiles(filePath, segmentPath);
const ok = await concatWebmFiles(filePath, segmentPath);
if (!ok) {
if (!ok) {
+ failed++;
console.warn(`[VOD] Failed to merge segment into ${path.basename(filePath)}: ${path.basename(segmentPath)}`);
console.warn(`[VOD] Failed to merge segment into ${path.basename(filePath)}: ${path.basename(segmentPath)}`);
}
}
}
}
+ return failed;
}
}
module.exports = {
module.exports = {
remuxForSeeking,
remuxForSeeking,
+ remuxForSeekingDetailed,
+ lastProgressSeconds,
+ probeDuration,
+ streamCopyDuration,
+ chooseDuration,
remuxForLiveSeeking,
remuxForLiveSeeking,
seekableSidecarPath,
seekableSidecarPath,
cleanupSeekableFile,
cleanupSeekableFile,
@@ -261,7 +261,8 @@ class StreamRecorder {
@@ -261,7 +261,8 @@ class StreamRecorder {
require('./finalize').finalizeVod(vodId, { startTimeMs: rec?.startTime }).catch(() => {
require('./finalize').finalizeVod(vodId, { startTimeMs: rec?.startTime }).catch(() => {
// Never bare-mark a failed recording as ready — that published
// Never bare-mark a failed recording as ready — that published
// 0:00 ghosts. Quarantine it out of listings instead.
// 0:00 ghosts. Quarantine it out of listings instead.
- db.run(`UPDATE vods SET is_recording = 0, health_status = 'needs_review',
+ // The live duration updates were wall-clock estimates: none of it is kept.
+ db.run(`UPDATE vods SET is_recording = 0, duration_seconds = 0, duration_source = 'unknown', health_status = 'needs_review',
health_issues_json = ?, quarantined_at = datetime('now'), is_public = 0 WHERE id = ?`,
health_issues_json = ?, quarantined_at = datetime('now'), is_public = 0 WHERE id = ?`,
[JSON.stringify(['finalize_failed']), vodId]);
[JSON.stringify(['finalize_failed']), vodId]);
});
});
@@ -327,7 +327,7 @@ router.get('/:id', tenantAuth({ allowUser: true }), async (req, res) => {
@@ -327,7 +327,7 @@ router.get('/:id', tenantAuth({ allowUser: true }), async (req, res) => {
const duration = await tools.probeVodDuration(vod.file_path);
const duration = await tools.probeVodDuration(vod.file_path);
if (duration > 0) {
if (duration > 0) {
const fileSize = tools.getFileSizeSafe(vod.file_path);
const fileSize = tools.getFileSizeSafe(vod.file_path);
- db.run('UPDATE vods SET duration_seconds = ?, file_size = ? WHERE id = ?', [duration, fileSize, vod.id]);
+ db.run("UPDATE vods SET duration_seconds = ?, file_size = ?, duration_source = 'probe' WHERE id = ?", [duration, fileSize, vod.id]);
vod.duration_seconds = duration;
vod.duration_seconds = duration;
vod.file_size = fileSize;
vod.file_size = fileSize;
}
}
new file mode 100644
new file mode 100644
@@ -0,0 +1,184 @@
@@ -0,0 +1,184 @@
+'use strict';
+// VOD duration truthfulness (server/vod/finalize.js, media-tools.chooseDuration): the stored duration
+// is what the file says (ffprobe, else the packet timeline of a stream-copy pass), never the wall
+// clock; vods.duration_source says which (probe | remux | unknown). Nothing measurable stores 0 with
+// needs_review; an orphan (no recorder start time) is never measured as now - created_at; a later
+// good finalize lifts the quarantine a failed one left.
+const assert = require('assert');
+const fs = require('fs');
+const os = require('os');
+const path = require('path');
+const { spawnSync } = require('child_process');
+
+const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-duration-'));
+const dir = (n) => { const d = path.join(tmp, n); fs.mkdirSync(d, { recursive: true }); return d; };
+process.env.DB_PATH = path.join(tmp, 'media.db');
+process.env.VOD_PATH = dir('vods');
+process.env.CLIPS_PATH = dir('clips');
+process.env.THUMBNAILS_PATH = dir('thumbnails');
+process.env.FILES_PATH = dir('files');
+process.env.PASTES_PATH = dir('pastes');
+process.env.OBJECTS_PATH = dir('objects');
+process.env.MEDIA_JOBS_ENABLED = '0';
+
+const tools = require('../server/vod/media-tools');
+
+// ── 1. chooseDuration: measured values only, bounded by the recorder's run when known ──
+{
+ const c = tools.chooseDuration;
+ assert.deepStrictEqual(c({ probeS: 600.4 }), { seconds: 600.4, source: 'probe', issues: [] });
+ assert.deepStrictEqual(c({ probeS: 0, remuxS: 590 }), { seconds: 590, source: 'remux', issues: [] }, 'no container duration: the packets');
+ assert.deepStrictEqual(c({}), { seconds: 0, source: 'unknown', issues: ['probe_failed'] }, 'nothing measured: 0, unknown');
+ const inflated = c({ probeS: 86400, remuxS: 600 });
+ assert.deepStrictEqual([inflated.seconds, inflated.source], [600, 'remux'], 'a header far longer than the packets is corrupt');
+ assert.ok(inflated.issues.includes('inflated_container_duration'));
+ const bounded = c({ probeS: 5000, remuxS: 0, wallS: 600 });
+ assert.deepStrictEqual([bounded.seconds, bounded.source], [0, 'unknown'], 'longer than the recorder ran: refused, and the wall clock is NOT used');
+ assert.ok(bounded.issues.includes('inflated_duration'));
+ assert.strictEqual(c({ probeS: 5000, remuxS: 590, wallS: 600 }).source, 'remux', 'the plausible measurement wins');
+ assert.strictEqual(c({ probeS: 172800 }).seconds, 172800, 'no start time (orphan): no bound either, only measurements');
+ assert.strictEqual(tools.lastProgressSeconds('frame=1 time=00:00:01.00 x\rframe=2 time=01:02:03.50 bitrate'), 3723.5);
+ assert.strictEqual(tools.lastProgressSeconds('time=N/A'), 0);
+ console.log('✅ chooseDuration: probe, else the packet timeline, else 0/unknown; bounded by the recorder run; never the wall clock');
+}
+
+const db = require('../server/db/database');
+const finalize = require('../server/vod/finalize');
+db.upsertApp({ app_id: 'live', api_key: 'live-key' });
+
+const cols = db.all('PRAGMA table_info(vods)').map(c => c.name);
+assert.ok(cols.includes('duration_source'), 'vods.duration_source exists');
+
+const daysAgo = (d) => new Date(Date.now() - d * 864e5).toISOString().replace('T', ' ').slice(0, 19);
+function makeVod(file, { createdAt = daysAgo(2), wallDuration = 99999 } = {}) {
+ const r = db.run(`INSERT INTO vods (app_id, user_id, title, file_path, is_recording, duration_seconds, created_at, is_public, visibility)
+ VALUES ('live', 7, 'rec', ?, 1, ?, ?, 1, 'public')`, [file, wallDuration, createdAt]);
+ return Number(r.lastInsertRowid);
+}
+const row = (id) => db.get('SELECT * FROM vods WHERE id = ?', [id]);
+
+const realTools = { ...tools };
+function restore() { Object.assign(tools, realTools); }
+
+(async () => {
+ // ── 2. Probe failure: 0 + needs_review + hidden; the live wall-clock estimate is not kept ──
+ {
+ const file = path.join(process.env.VOD_PATH, 'garbage.mp4');
+ fs.writeFileSync(file, Buffer.alloc(20000, 7));
+ const id = makeVod(file);
+ tools.remuxForSeekingDetailed = async () => ({ ok: false, seconds: 0, error: 'ffmpeg remux exited 1' });
+ tools.probeDuration = async () => ({ ok: false, seconds: 0, format: null, streams: [], error: 'unreadable' });
+ tools.streamCopyDuration = async () => ({ ok: false, seconds: 0, error: 'no packets' });
+ await finalize.finalizeVod(id);
+ const v = row(id);
+ assert.strictEqual(v.duration_seconds, 0, 'probe failure stores 0, not the recorder\'s wall-clock estimate');
+ assert.strictEqual(v.duration_source, 'unknown');
+ assert.strictEqual(v.health_status, 'needs_review');
+ const issues = JSON.parse(v.health_issues_json);
+ assert.ok(issues.includes('probe_failed') && issues.includes('remux_failed'), `issues: ${issues}`);
+ assert.strictEqual(v.is_recording, 0);
+ assert.strictEqual(v.is_public, 0, 'hidden until reviewed');
+ assert.ok(v.quarantined_at);
+
+ // A later finalize that can measure it lifts the quarantine (the vod.finalize retry job does this).
+ tools.probeDuration = async () => ({ ok: true, seconds: 1234.4, format: { format_name: 'mp4' }, streams: [], error: null });
+ await finalize.finalizeVod(id);
+ const w = row(id);
+ assert.deepStrictEqual([w.duration_seconds, w.duration_source, w.health_status], [1234, 'probe', 'ok']);
+ assert.strictEqual(w.quarantined_at, null, 'quarantine lifted');
+ assert.strictEqual(w.is_public, 1, 'back to the visibility its owner chose');
+ restore();
+ console.log('✅ probe failure: duration 0, duration_source unknown, needs_review, hidden; a later good finalize lifts it');
+ }
+
+ // ── 3. Orphans (no start time) never use now - created_at; a bound applies only with a real start ──
+ {
+ const file = path.join(process.env.VOD_PATH, 'orphan.mp4');
+ fs.writeFileSync(file, Buffer.alloc(20000, 1));
+ const id = makeVod(file, { createdAt: daysAgo(3) });
+ tools.remuxForSeekingDetailed = async () => ({ ok: true, seconds: 3601.2, error: null });
+ tools.probeDuration = async () => ({ ok: true, seconds: 3600.8, format: {}, streams: [], error: null });
+ await finalize.finalizeVod(id);
+ const v = row(id);
+ assert.deepStrictEqual([v.duration_seconds, v.duration_source], [3601, 'probe'], 'orphan: the measured length, not 3 days');
+
+ const file2 = path.join(process.env.VOD_PATH, 'inflated.mp4');
+ fs.writeFileSync(file2, Buffer.alloc(20000, 2));
+ const id2 = makeVod(file2);
+ tools.probeDuration = async () => ({ ok: true, seconds: 50000, format: {}, streams: [], error: null });
+ tools.remuxForSeekingDetailed = async () => ({ ok: true, seconds: 50000, error: null });
+ await finalize.finalizeVod(id2, { startTimeMs: Date.now() - 600 * 1000 });
+ const w = row(id2);
+ assert.deepStrictEqual([w.duration_seconds, w.duration_source, w.health_status], [0, 'unknown', 'needs_review'],
+ 'longer than the recorder ran: 0 and review, where the old code stored the wall clock');
+ assert.ok(JSON.parse(w.health_issues_json).includes('inflated_duration'));
+
+ const file3 = path.join(process.env.VOD_PATH, 'header.mp4');
+ fs.writeFileSync(file3, Buffer.alloc(20000, 3));
+ const id3 = makeVod(file3);
+ tools.probeDuration = async () => ({ ok: true, seconds: 90000, format: {}, streams: [], error: null });
+ tools.remuxForSeekingDetailed = async () => ({ ok: true, seconds: 581, error: null });
+ await finalize.finalizeVod(id3, { startTimeMs: Date.now() - 600 * 1000 });
+ assert.deepStrictEqual([row(id3).duration_seconds, row(id3).duration_source, row(id3).health_status], [581, 'remux', 'ok'], 'the packet timeline when the header lies');
+ restore();
+ console.log('✅ orphans are measured, never now - created_at; impossible values need review; a lying header yields to the packets');
+ }
+
+ // ── 4. Short, corrupt and stat-failure paths also drop the wall-clock estimate ──
+ {
+ const file = path.join(process.env.VOD_PATH, 'short.mp4');
+ fs.writeFileSync(file, Buffer.alloc(20000, 4));
+ const id = makeVod(file);
+ tools.remuxForSeekingDetailed = async () => ({ ok: true, seconds: 1, error: null });
+ tools.probeDuration = async () => ({ ok: true, seconds: 1, format: {}, streams: [], error: null });
+ await finalize.finalizeVod(id);
+ assert.deepStrictEqual([row(id).duration_seconds, row(id).duration_source, row(id).health_status], [1, 'probe', 'needs_review']);
+
+ const file2 = path.join(process.env.VOD_PATH, 'corrupt.mp4');
+ fs.writeFileSync(file2, Buffer.alloc(20000, 5));
+ const id2 = makeVod(file2);
+ tools.probeDuration = async () => ({ ok: true, seconds: 42, format: {}, streams: [], error: null });
+ tools.remuxForSeekingDetailed = async () => ({ ok: true, seconds: 42, error: null });
+ await finalize.finalizeVod(id2, { ffmpegCorrupted: true });
+ assert.deepStrictEqual([row(id2).duration_seconds, row(id2).duration_source, row(id2).health_status], [42, 'probe', 'corrupt'],
+ 'the corrupt path stores the measurement, not the live estimate');
+
+ const file3 = path.join(process.env.VOD_PATH, 'vanishing.mp4');
+ fs.writeFileSync(file3, Buffer.alloc(20000, 6));
+ const id3 = makeVod(file3);
+ tools.probeDuration = async (p) => { if (p === file3) { try { fs.unlinkSync(file3); } catch { /* */ } } return { ok: true, seconds: 300, format: {}, streams: [], error: null }; };
+ tools.remuxForSeekingDetailed = async () => ({ ok: true, seconds: 300, error: null });
+ await finalize.finalizeVod(id3);
+ const v3 = row(id3);
+ assert.deepStrictEqual([v3.is_recording, v3.duration_seconds, v3.duration_source, v3.health_status], [0, 0, 'unknown', 'needs_review']);
+ assert.ok(JSON.parse(v3.health_issues_json).includes('stat_failed'));
+ restore();
+ console.log('✅ short, corrupt and stat-failure outcomes never keep the wall-clock estimate');
+ }
+
+ // ── 5. Real ffmpeg: an orphaned 6 s recording made 2 days ago finalizes to ~6 s (probe) ──
+ if (spawnSync('ffmpeg', ['-version']).status !== 0) {
+ console.log('⚠️ ffmpeg not found: the real-file finalize check is skipped');
+ } else {
+ const file = path.join(process.env.VOD_PATH, 'real.mp4');
+ const mk = spawnSync('ffmpeg', ['-v', 'error', '-y', '-f', 'lavfi', '-i', 'testsrc=duration=6:size=160x120:rate=10',
+ '-f', 'lavfi', '-i', 'sine=frequency=440:duration=6', '-c:v', 'libx264', '-pix_fmt', 'yuv420p', '-c:a', 'aac', '-shortest',
+ '-movflags', '+frag_keyframe+empty_moov+default_base_moof', '-f', 'mp4', file]);
+ assert.strictEqual(mk.status, 0, String(mk.stderr));
+ const id = makeVod(file, { createdAt: daysAgo(2) });
+ await finalize.finalizeVod(id);
+ const v = row(id);
+ assert.ok(v.duration_seconds >= 5 && v.duration_seconds <= 7, `measured ${v.duration_seconds}s`);
+ assert.strictEqual(v.duration_source, 'probe');
+ assert.strictEqual(v.health_status, 'ok');
+ const packets = await tools.streamCopyDuration(file);
+ assert.ok(packets.ok && packets.seconds > 5 && packets.seconds < 7, `stream-copy pass measured ${packets.seconds}`);
+ const pr = await tools.probeDuration(path.join(tmp, 'nope.mp4'));
+ assert.strictEqual(pr.ok, false, 'an unreadable source is a failed probe, not a zero-length one');
+ console.log('✅ a real orphaned recording finalizes to its measured length (probe), and the packet pass agrees');
+ }
+
+ db.close();
+ fs.rmSync(tmp, { recursive: true, force: true });
+ console.log('\n✅ All VOD duration tests passed');
+})().catch((err) => { console.error(err); process.exit(1); });