Cancel stale playback probes and verify Linux QSV support
This commit is contained in:
Technomancer702
@ -1,3 +1,5 @@
|
||||
const { ProbeLeases, parseProbeLease, requestCancellation, cancelled } = require('../services/probeLeases');
|
||||
const probeLeases = new ProbeLeases();
|
||||
const { parseProgram, selectProgramStreams } = require('../services/mpegTsProgram');
|
||||
const { redactUrl, redactText, validateHttpUrl, FFMPEG_PROTOCOL_WHITELIST } = require('../services/urlSecurity');
|
||||
const express = require('express');
|
||||
@ -30,7 +32,7 @@ const BROWSER_AUDIO_CODECS = ['aac', 'mp3', 'opus', 'vorbis'];
|
||||
/**
|
||||
* Probe stream with ffprobe
|
||||
*/
|
||||
function probeStream(url, ffprobePath, userAgent = null, timeout = 15000) {
|
||||
function probeStream(url, ffprobePath, userAgent = null, timeout = 15000, signal) {
|
||||
return new Promise((resolve, reject) => {
|
||||
const args = [
|
||||
'-v', 'error',
|
||||
@ -50,35 +52,28 @@ function probeStream(url, ffprobePath, userAgent = null, timeout = 15000) {
|
||||
url
|
||||
];
|
||||
|
||||
const proc = spawn(ffprobePath, args);
|
||||
let stdout = '';
|
||||
let stderr = '';
|
||||
|
||||
const timer = setTimeout(() => {
|
||||
if (signal?.aborted) { reject(cancelled()); return; }
|
||||
const proc = spawn(ffprobePath, args, { windowsHide: true });
|
||||
let stdout = '', stderr = '', failure;
|
||||
const stop = error => {
|
||||
if (failure) return;
|
||||
failure = error;
|
||||
proc.kill('SIGKILL');
|
||||
reject(new Error('Probe timeout'));
|
||||
}, timeout);
|
||||
|
||||
proc.stdout.on('data', (data) => { stdout += data; });
|
||||
proc.stderr.on('data', (data) => { stderr += data; });
|
||||
|
||||
proc.on('close', (code) => {
|
||||
};
|
||||
const abort = () => stop(cancelled());
|
||||
const timer = setTimeout(() => stop(new Error('Probe timeout')), timeout);
|
||||
signal?.addEventListener('abort', abort, { once: true });
|
||||
if (signal?.aborted) abort();
|
||||
proc.stdout.on('data', data => { stdout += data; });
|
||||
proc.stderr.on('data', data => { stderr = (stderr + data).slice(-65536); });
|
||||
proc.on('error', error => { failure ||= error; });
|
||||
proc.once('close', code => {
|
||||
clearTimeout(timer);
|
||||
if (code !== 0) {
|
||||
reject(new Error(`ffprobe exited with code ${code}: ${stderr}`));
|
||||
return;
|
||||
}
|
||||
try {
|
||||
const result = JSON.parse(stdout);
|
||||
resolve(result);
|
||||
} catch (e) {
|
||||
reject(new Error('Failed to parse ffprobe output'));
|
||||
}
|
||||
});
|
||||
|
||||
proc.on('error', (err) => {
|
||||
clearTimeout(timer);
|
||||
reject(err);
|
||||
signal?.removeEventListener('abort', abort);
|
||||
if (failure) { reject(failure); return; }
|
||||
if (code !== 0) { reject(new Error(`ffprobe exited with code ${code}: ${redactText(stderr)}`)); return; }
|
||||
try { resolve(JSON.parse(stdout)); }
|
||||
catch { reject(new Error('Failed to parse ffprobe output')); }
|
||||
});
|
||||
});
|
||||
}
|
||||
@ -154,68 +149,45 @@ function analyzeProbeResult(probeResult, url, program = null) {
|
||||
|
||||
router.get('/', async (req, res) => {
|
||||
const { url, ua } = req.query;
|
||||
let program;
|
||||
try { program = parseProgram(req.query.program); } catch (error) { return res.status(400).json({ error: error.message }); }
|
||||
try { validateHttpUrl(url); } catch (error) { return res.status(400).json({ error: error.message }); }
|
||||
if (!url) {
|
||||
return res.status(400).json({ error: 'URL parameter is required' });
|
||||
}
|
||||
|
||||
const ffprobePath = req.app.locals.ffprobePath;
|
||||
const cacheKey = JSON.stringify([url, ua || '', program]);
|
||||
|
||||
if (!ffprobePath) {
|
||||
// No ffprobe available - assume needs transcoding to be safe
|
||||
console.log('[Probe] FFprobe not available, assuming transcode needed');
|
||||
return res.json({
|
||||
video: 'unknown',
|
||||
audio: 'unknown',
|
||||
container: 'unknown',
|
||||
compatible: false,
|
||||
needsRemux: false,
|
||||
needsTranscode: true
|
||||
});
|
||||
}
|
||||
|
||||
// Check cache
|
||||
const cached = probeCache.get(cacheKey);
|
||||
if (cached && (Date.now() - cached.timestamp < CACHE_TTL)) {
|
||||
console.log(`[Probe] Cache hit for: ${redactUrl(url)}...`);
|
||||
return res.json(cached.result);
|
||||
}
|
||||
|
||||
console.log(`[Probe] Probing: ${redactUrl(url)}... ${ua ? `(UA: ${ua})` : ''}`);
|
||||
|
||||
let program, lease;
|
||||
try {
|
||||
const probeResult = await probeStream(url, ffprobePath, ua);
|
||||
const analysis = analyzeProbeResult(probeResult, url, program);
|
||||
program = parseProgram(req.query.program);
|
||||
lease = parseProbeLease(req.query);
|
||||
validateHttpUrl(url);
|
||||
if (ua !== undefined && (typeof ua !== 'string' || ua.length > 1024 || /[\x00-\x1f\x7f]/.test(ua))) throw new Error('Invalid probe user agent');
|
||||
} catch (error) { return res.status(400).json({ error: error.message }); }
|
||||
|
||||
// Cache result
|
||||
probeCache.set(cacheKey, { result: analysis, timestamp: Date.now() });
|
||||
|
||||
console.log(`[Probe] Result: video=${analysis.video}, audio=${analysis.audio}, ` +
|
||||
`container=${analysis.container}, compatible=${analysis.compatible}, ` +
|
||||
`needsRemux=${analysis.needsRemux}, needsTranscode=${analysis.needsTranscode}`);
|
||||
|
||||
res.json(analysis);
|
||||
} catch (err) {
|
||||
console.error('[Probe] Failed:', err.message);
|
||||
|
||||
// On error, assume transcode needed to be safe
|
||||
const fallback = {
|
||||
video: 'unknown',
|
||||
audio: 'unknown',
|
||||
container: 'unknown',
|
||||
compatible: false,
|
||||
needsRemux: false,
|
||||
needsTranscode: true,
|
||||
error: err.message
|
||||
};
|
||||
|
||||
probeCache.set(cacheKey, { result: fallback, timestamp: Date.now() - (CACHE_TTL - 30000) });
|
||||
|
||||
res.json(fallback);
|
||||
}
|
||||
const cancellation = requestCancellation(req, res);
|
||||
const cacheKey = JSON.stringify([url, ua || '', program]);
|
||||
try {
|
||||
const result = await probeLeases.run(req.user?.id ?? null, lease, cancellation.signal, async signal => {
|
||||
const ffprobePath = req.app.locals.ffprobePath;
|
||||
if (!ffprobePath) return { video: 'unknown', audio: 'unknown', container: 'unknown', compatible: false, needsRemux: false, needsTranscode: true };
|
||||
const cached = probeCache.get(cacheKey);
|
||||
if (cached && Date.now() - cached.timestamp < CACHE_TTL) return cached.result;
|
||||
console.log(`[Probe] Probing: ${redactUrl(url)}...`);
|
||||
try {
|
||||
const data = await probeStream(url, ffprobePath, ua, req.app.locals.probeTimeoutMs || 15000, signal);
|
||||
if (signal.aborted) throw cancelled();
|
||||
const analysis = analyzeProbeResult(data, url, program);
|
||||
probeCache.set(cacheKey, { result: analysis, timestamp: Date.now() });
|
||||
return analysis;
|
||||
} catch (error) {
|
||||
if (signal.aborted || error.name === 'AbortError') throw cancelled();
|
||||
const message = redactText(error.message);
|
||||
console.error('[Probe] Failed:', message);
|
||||
const fallback = { video: 'unknown', audio: 'unknown', container: 'unknown', compatible: false,
|
||||
needsRemux: false, needsTranscode: true, error: message };
|
||||
probeCache.set(cacheKey, { result: fallback, timestamp: Date.now() - (CACHE_TTL - 30000) });
|
||||
return fallback;
|
||||
}
|
||||
});
|
||||
if (!cancellation.signal.aborted && !res.destroyed) res.json(result);
|
||||
} catch (error) {
|
||||
if (cancellation.signal.aborted || res.destroyed) return;
|
||||
if (error.name === 'AbortError') return res.status(409).json({ error: 'Playback probe was superseded' });
|
||||
res.status(error.code === 'PROBE_LIMIT' ? 503 : 500).json({ error: 'Unable to probe stream' });
|
||||
} finally { cancellation.dispose(); }
|
||||
});
|
||||
|
||||
module.exports = router;
|
||||
|
||||
Reference in New Issue
Block a user