diff --git a/server/src/services/replayEngineV2.js b/server/src/services/replayEngineV2.js index 16e3eb63..bb007db9 100644 --- a/server/src/services/replayEngineV2.js +++ b/server/src/services/replayEngineV2.js @@ -26,9 +26,11 @@ const MAX_BYTES = Math.floor(Number.parseFloat(process.env.REPLAY_MAX_OUTPUT_MB const events = new EventEmitter(); const workers = new Map(); // key -> worker +const pendingWorkerStarts = new Set(); // key const segmentIndex = new Map(); // key -> [{filePath,startMs,endMs,mtimeMs,size,kind,sourceType,sourceId}] let cleanupTimer = null; let activeSegmentRoot = SEGMENT_ROOT; +let tickInFlight = false; function sourceKey(source) { return `${source.sourceType}__${source.kind}__${source.id}`; @@ -168,10 +170,15 @@ async function ensureDir(dir) { function startWorker(source) { const key = sourceKey(source); - if (workers.has(key)) return; + if (workers.has(key) || pendingWorkerStarts.has(key)) return; + pendingWorkerStarts.add(key); const dir = sourceDirForKey(key); ensureDir(dir) .then(() => { + if (workers.has(key)) { + pendingWorkerStarts.delete(key); + return; + } const proc = spawn(FFMPEG_BIN, buildWorkerArgs(source), { stdio: ['ignore', 'ignore', 'pipe'] }); const worker = { key, @@ -179,6 +186,7 @@ function startWorker(source) { proc, }; workers.set(key, worker); + pendingWorkerStarts.delete(key); proc.stderr.on('data', (chunk) => { const text = String(chunk || '').trim(); if (!text) return; @@ -199,6 +207,7 @@ function startWorker(source) { }); }) .catch((err) => { + pendingWorkerStarts.delete(key); logger.warn('failed to start worker', { key, error: err.message }); }); } @@ -232,27 +241,6 @@ async function syncWorkers() { } } -async function probeDurationSec(filePath) { - try { - const { stdout } = await execFileAsync('ffprobe', [ - '-v', - 'error', - '-show_entries', - 'format=duration', - '-of', - 'default=noprint_wrappers=1:nokey=1', - filePath, - ]); - const value = Number(String(stdout || '').trim()); - if (Number.isFinite(value) && value > 0 && value < SEGMENT_SECONDS * 10) { - return value; - } - } catch { - // ignore probe failures; fallback below - } - return SEGMENT_SECONDS; -} - async function refreshIndexForWorker(worker) { const key = worker.key; const dir = sourceDirForKey(key); @@ -283,7 +271,7 @@ async function refreshIndexForWorker(worker) { continue; } if (!stat.isFile() || stat.size < 4096) continue; - const durationSec = await probeDurationSec(filePath); + const durationSec = SEGMENT_SECONDS; const endMs = Math.round(stat.mtimeMs); const startMs = Math.round(endMs - durationSec * 1000); if (endMs < cutoffMs) continue; @@ -664,10 +652,16 @@ async function bootstrapIndexFromDisk() { } async function tick() { + if (tickInFlight) return; + tickInFlight = true; + try { await syncWorkers(); await refreshSegmentIndex(); await cleanupOldFiles(); events.emit('health', getReplayHealthSnapshot()); + } finally { + tickInFlight = false; + } } async function start() {