diff --git a/public/index.html b/public/index.html index 6a4ef65..b8864fa 100644 --- a/public/index.html +++ b/public/index.html @@ -1097,8 +1097,9 @@ + - + @@ -1108,7 +1109,7 @@ - + diff --git a/public/js/app.js b/public/js/app.js index 5e8b4bd..ee8c63d 100644 --- a/public/js/app.js +++ b/public/js/app.js @@ -220,6 +220,8 @@ class App { logoutLink.addEventListener('click', async (e) => { e.preventDefault(); + this.player.stop(); + this.pages.watch.stop(); await fetch('/api/auth/logout', { method: 'POST', credentials: 'same-origin' }); localStorage.removeItem('authToken'); diff --git a/public/js/components/PlaybackSessions.js b/public/js/components/PlaybackSessions.js new file mode 100644 index 0000000..ec1a90e --- /dev/null +++ b/public/js/components/PlaybackSessions.js @@ -0,0 +1,52 @@ +(function (root) { + const cancelled = () => Object.assign(new Error('Playback was superseded'), { name: 'AbortError' }); + class PlaybackSessions { + constructor(isLive) { + this.isLive = isLive; + this.leaseId = null; + this.sequence = 0; + this.controller = null; + this.timer = null; + root.addEventListener?.('pagehide', () => this.stop()); + } + async create(url, options) { + this.leaseId ||= root.crypto.randomUUID?.() || Array.from(root.crypto.getRandomValues(new Uint8Array(16)), + byte => byte.toString(16).padStart(2, '0')).join(''); + const leaseId = this.leaseId, sequence = ++this.sequence; + this.controller?.abort(); + const controller = this.controller = new AbortController(); + const response = await fetch('/api/transcode/session', { + method: 'POST', credentials: 'same-origin', signal: controller.signal, + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ ...options, url, isLive: this.isLive, playbackLeaseId: leaseId, requestSequence: sequence }) + }); + const session = await response.json(); + if (controller.signal.aborted || leaseId !== this.leaseId || sequence !== this.sequence) { + if (session.sessionId) fetch(`/api/transcode/${session.sessionId}`, { method: 'DELETE', keepalive: true }).catch(() => {}); + throw cancelled(); + } + if (!response.ok) throw new Error(session.error || 'Unable to start playback'); + this.timer ||= setInterval(() => { + if (this.leaseId) fetch('/api/transcode/lease/heartbeat', { + method: 'POST', headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ playbackLeaseId: this.leaseId }) + }).catch(() => {}); + }, 30000); + return session; + } + stop() { + const leaseId = this.leaseId; + this.leaseId = null; + this.sequence++; + this.controller?.abort(); + this.controller = null; + clearInterval(this.timer); + this.timer = null; + if (leaseId) fetch('/api/transcode/lease', { + method: 'DELETE', keepalive: true, headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ playbackLeaseId: leaseId }) + }).catch(() => {}); + } + } + root.PlaybackSessions = PlaybackSessions; +}(typeof window === 'object' ? window : globalThis)); diff --git a/public/js/components/VideoPlayer.js b/public/js/components/VideoPlayer.js index 076018c..6e42fb8 100644 --- a/public/js/components/VideoPlayer.js +++ b/public/js/components/VideoPlayer.js @@ -813,38 +813,29 @@ class VideoPlayer { * Start a HLS transcode session */ async startTranscodeSession(url, options = {}) { + const generation = this.playbackGeneration; + this.playbackSessions ||= new PlaybackSessions(true); + const sessionOptions = options; try { - console.log('[Player] Starting HLS transcode session...', options); - const res = await fetch('/api/transcode/session', { - method: 'POST', - headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ url, ...options }) - }); - if (!res.ok) throw new Error('Failed to start session'); - const session = await res.json(); + const session = await this.playbackSessions.create(url, sessionOptions); + if (generation !== this.playbackGeneration) return null; this.currentSessionId = session.sessionId; + return session.playlistUrl; - } catch (err) { - console.error('[Player] Session start failed:', err); - // Fallback to direct transcode if session fails - return `/api/transcode?url=${encodeURIComponent(url)}`; + } catch (error) { + if (error.name !== 'AbortError' && generation === this.playbackGeneration) { + this.stopTranscodeSession(); + console.error('Unable to start playback:', error.message); + this.showError('Unable to start transcoding. Please try again.'); + } + return null; } } - /** - * Stop and cleanup current transcode session - */ - async stopTranscodeSession() { - if (this.currentSessionId) { - console.log('[Player] Stopping transcode session:', this.currentSessionId); - try { - // Fire and forget cleanup - fetch(`/api/transcode/${this.currentSessionId}`, { method: 'DELETE' }); - } catch (err) { - console.error('Failed to stop session:', err); - } - this.currentSessionId = null; - } + stopTranscodeSession() { + this.playbackSessions?.stop(); + this.currentSessionId = null; + } /** @@ -920,6 +911,7 @@ class VideoPlayer { const statusMode = this.settings.upscaleEnabled ? 'upscaling' : 'transcoding'; this.updateTranscodeStatus(statusMode, statusText); + if (this.playbackGeneration !== playbackGeneration) return; const playlistUrl = await this.startTranscodeSession(streamUrl, { videoMode, videoCodec: info.video, @@ -927,6 +919,7 @@ class VideoPlayer { audioChannels: info.audioChannels, videoHeight: info.height }); + if (!playlistUrl || this.playbackGeneration !== playbackGeneration) return; this.currentUrl = playlistUrl; // Update currentUrl for HLS reload this.playHls(playlistUrl); @@ -960,48 +953,20 @@ class VideoPlayer { } } + if (this.playbackGeneration !== playbackGeneration) return; // CHECK: Force Video Transcode (Full) or Upscaling if (this.settings.forceVideoTranscode || this.settings.upscaleEnabled) { const statusText = this.settings.upscaleEnabled ? 'Upscaling' : 'Transcoding (Video)'; const statusMode = this.settings.upscaleEnabled ? 'upscaling' : 'transcoding'; console.log(`[Player] ${statusText} enabled. Starting session (encode)...`); this.updateTranscodeStatus(statusMode, statusText); + if (this.playbackGeneration !== playbackGeneration) return; const playlistUrl = await this.startTranscodeSession(streamUrl, { videoMode: 'encode' }); + if (!playlistUrl || this.playbackGeneration !== playbackGeneration) return; this.currentUrl = playlistUrl; - // Load HLS - this.updateNowPlaying(channel, 'Transcoding (Video)'); - // ... (rest is same logic flow, simplified by just falling through to playHls call if I refactored) - // But for minimize drift, I'll copy the block logic for HLS playback init - // Actually, I can just fall through if I set looksLikeHls = true? - // No, play logic is sequential. - if (Hls.isSupported()) { - // Start HLS - // ... this repeats code. I should probably just set currentUrl and let HLS block handle? - // But HLS block is lower down. - // I will just execute the HLS init here as before. - - // Actually, easiest way is to re-assign streamUrl and goto start? No. - // Copy existing forceTranscode block logic - if (this.hls) { - this.hls.destroy(); - } - this.hls = new Hls(); - this.hls.loadSource(playlistUrl); - this.hls.attachMedia(this.video); - this.hls.on(Hls.Events.MANIFEST_PARSED, () => { - this.video.play().catch(console.error); - }); - // Handle errors - this.hls.on(Hls.Events.ERROR, (event, data) => { - if (data.fatal) { - console.log('[Player] HLS fatal error'); - this.hls.destroy(); - } - }); - - return; // Exit - } + this.playHls(playlistUrl); + return; } // CHECK: Force Audio Transcode (Copy Video) - legacy forceTranscode setting @@ -1014,10 +979,13 @@ class VideoPlayer { try { const probeRes = await fetch(`/api/probe?url=${encodeURIComponent(streamUrl)}`); const info = await probeRes.json(); - videoCodec = info.video; + if (this.playbackGeneration !== playbackGeneration) return; + videoCodec = info.video; } catch (e) { console.warn('Probe failed for force audio, assuming h264'); } + if (this.playbackGeneration !== playbackGeneration) return; const playlistUrl = await this.startTranscodeSession(streamUrl, { videoMode: 'copy', videoCodec }); + if (!playlistUrl || this.playbackGeneration !== playbackGeneration) return; this.currentUrl = playlistUrl; console.log('[Player] Playing transcoded HLS stream:', playlistUrl); @@ -1210,20 +1178,24 @@ class VideoPlayer { } this.hls = new Hls(this.getHlsConfig()); + const hls = this.hls, generation = this.playbackGeneration; this.hls.loadSource(url); this.hls.attachMedia(this.video); this.hls.on(Hls.Events.MANIFEST_PARSED, () => { + if (this.hls !== hls || this.playbackGeneration !== generation) return; this.video.play().catch(e => { if (e.name !== 'AbortError') console.log('Autoplay prevented:', e); }); }); this.hls.on(Hls.Events.ERROR, (event, data) => { + if (this.hls !== hls || this.playbackGeneration !== generation) return; if (data.fatal) { // Simple error handling for forced HLS/transcode modes console.error('Fatal HLS error in transcode mode:', data); this.hls.destroy(); + this.stopTranscodeSession(); } }); } diff --git a/public/js/pages/WatchPage.js b/public/js/pages/WatchPage.js index c07132b..a6340f8 100644 --- a/public/js/pages/WatchPage.js +++ b/public/js/pages/WatchPage.js @@ -180,7 +180,14 @@ class WatchPage { }); // Progress bar - this.progressSlider?.addEventListener('input', (e) => this.seek(e.target.value)); + this.progressSlider?.addEventListener('input', e => { + this.progressScrubbing = true; + this.timeCurrent.textContent = this.formatTime(this.getPlaybackDuration() * Number(e.target.value) / 100); + }); + this.progressSlider?.addEventListener('change', e => { + this.progressScrubbing = false; + this.seek(e.target.value); + }); // Video events this.video?.addEventListener('timeupdate', () => this.updateProgress()); @@ -246,6 +253,7 @@ class WatchPage { * @param {string} streamUrl - Stream URL */ async play(content, streamUrl) { + this.stop(); this.sourceDuration = 0; this.content = content; this.contentType = content.type; @@ -274,7 +282,10 @@ class WatchPage { this.subtitleEl.textContent = content.subtitle || ''; // Load video - await this.loadVideo(streamUrl); + const loading = this.loadVideo(streamUrl); + const generation = this.playbackGeneration; + await loading; + if (this.playbackGeneration !== generation) return; // Show Now Playing indicator in navbar this.showNowPlaying(content.title); @@ -328,42 +339,32 @@ class WatchPage { * Start a HLS transcode session */ async startTranscodeSession(url, options = {}) { + const generation = this.playbackGeneration; + this.playbackSessions ||= new PlaybackSessions(false); + const sessionOptions = { seekOffset: this.resumeTime, ...options }; try { - console.log('[WatchPage] Starting HLS transcode session...', options); - const res = await fetch('/api/transcode/session', { - method: 'POST', - headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ - url, - seekOffset: this.resumeTime, // Pass resume point to backend - ...options - }) - }); - if (!res.ok) throw new Error('Failed to start session'); - const session = await res.json(); + const session = await this.playbackSessions.create(url, sessionOptions); + if (generation !== this.playbackGeneration) return null; this.currentSessionId = session.sessionId; + this.playbackTimeOffset = Number(sessionOptions.seekOffset) || 0; + this.resumeTime = 0; + this.currentTranscodeOptions = sessionOptions; return session.playlistUrl; - } catch (err) { - console.error('[WatchPage] Session start failed:', err); - // Fallback to direct transcode if session fails - return `/api/transcode?url=${encodeURIComponent(url)}`; + } catch (error) { + if (error.name !== 'AbortError' && generation === this.playbackGeneration) { + this.stopTranscodeSession(); + console.error('Unable to start playback:', error.message); + this.updateTranscodeStatus('error', 'Unable to start playback'); + this.loadingSpinner?.classList.remove('show'); + } + return null; } } - /** - * Stop and cleanup current transcode session - */ - async stopTranscodeSession() { - if (this.currentSessionId) { - console.log('[WatchPage] Stopping transcode session:', this.currentSessionId); - try { - // Fire and forget cleanup - fetch(`/api/transcode/${this.currentSessionId}`, { method: 'DELETE' }); - } catch (err) { - console.error('Failed to stop session:', err); - } - this.currentSessionId = null; - } + stopTranscodeSession() { + this.playbackSessions?.stop(); + this.currentSessionId = null; + this.playbackTimeOffset = 0; } async updateTranscodeStatus(mode, text) { @@ -432,6 +433,7 @@ class WatchPage { console.warn('Could not load settings'); } + if (this.playbackGeneration !== playbackGeneration) return; // Detect stream type const looksLikeHls = url.includes('.m3u8') || url.includes('m3u8'); const isRawTs = url.includes('.ts') && !url.includes('.m3u8'); @@ -463,6 +465,7 @@ class WatchPage { const statusMode = settings.upscaleEnabled ? 'upscaling' : 'transcoding'; this.updateTranscodeStatus(statusMode, statusText); + if (this.playbackGeneration !== playbackGeneration) return; const playlistUrl = await this.startTranscodeSession(url, { videoMode, seekOffset: this.resumeTime, // Ensure seekOffset is passed @@ -471,6 +474,7 @@ class WatchPage { audioChannels: info.audioChannels, videoHeight: info.height }); + if (!playlistUrl || this.playbackGeneration !== playbackGeneration) return; this.playHls(playlistUrl); this.setVolumeFromStorage(); return; @@ -495,16 +499,19 @@ class WatchPage { } } + if (this.playbackGeneration !== playbackGeneration) return; // Priority 1: Force Video Transcode (Full) or Upscaling if (settings.forceVideoTranscode || settings.upscaleEnabled) { const statusText = settings.upscaleEnabled ? 'Upscaling' : 'Transcoding (Video)'; const statusMode = settings.upscaleEnabled ? 'upscaling' : 'transcoding'; console.log(`[WatchPage] ${statusText} enabled. Starting session (encode)...`); this.updateTranscodeStatus(statusMode, statusText); + if (this.playbackGeneration !== playbackGeneration) return; const playlistUrl = await this.startTranscodeSession(url, { videoMode: 'encode', seekOffset: this.resumeTime }); + if (!playlistUrl || this.playbackGeneration !== playbackGeneration) return; this.playHls(playlistUrl); this.setVolumeFromStorage(); return; @@ -520,14 +527,17 @@ class WatchPage { const ua = settings.userAgentPreset === 'custom' ? settings.userAgentCustom : settings.userAgentPreset; const probeRes = await fetch(`/api/probe?url=${encodeURIComponent(url)}&ua=${encodeURIComponent(ua || '')}`); const info = await probeRes.json(); + if (this.playbackGeneration !== playbackGeneration) return; videoCodec = info.video; } catch (e) { console.warn('Probe failed for force audio, assuming h264'); } + if (this.playbackGeneration !== playbackGeneration) return; const playlistUrl = await this.startTranscodeSession(url, { videoMode: 'copy', videoCodec, seekOffset: this.resumeTime }); + if (!playlistUrl || this.playbackGeneration !== playbackGeneration) return; this.playHls(playlistUrl); this.setVolumeFromStorage(); return; @@ -594,27 +604,32 @@ class WatchPage { enableWorker: true, }); + const hls = this.hls, generation = this.playbackGeneration; this.hls.loadSource(url); this.hls.attachMedia(this.video); // Listen for subtitle track updates this.hls.on(Hls.Events.SUBTITLE_TRACKS_UPDATED, (event, data) => { + if (this.hls !== hls || this.playbackGeneration !== generation) return; console.log('[WatchPage] Subtitle tracks updated:', data.subtitleTracks); // Wait a moment for native text tracks to populate - setTimeout(() => this.updateCaptionsTracks(), 100); + setTimeout(() => { if (this.hls === hls && this.playbackGeneration === generation) this.updateCaptionsTracks(); }, 100); }); this.hls.on(Hls.Events.SUBTITLE_TRACK_SWITCH, (event, data) => { + if (this.hls !== hls || this.playbackGeneration !== generation) return; console.log('[WatchPage] Subtitle track switched:', data); }); this.hls.on(Hls.Events.MANIFEST_PARSED, () => { + if (this.hls !== hls || this.playbackGeneration !== generation) return; this.video.play().catch(e => { if (e.name !== 'AbortError') console.error('[WatchPage] Autoplay error:', e); }); }); this.hls.on(Hls.Events.ERROR, (event, data) => { + if (this.hls !== hls || this.playbackGeneration !== generation) return; if (data.fatal) { console.error('[WatchPage] HLS fatal error:', data); // Try proxy on CORS error (only if not already proxied/transcoded) @@ -624,6 +639,7 @@ class WatchPage { this.playHls(`/api/proxy/stream?url=${encodeURIComponent(this.currentUrl)}`); } else { this.hls.destroy(); + this.stopTranscodeSession(); } } }); @@ -646,6 +662,8 @@ class WatchPage { this.saveProgress(); this.sourceDuration = 0; this.durationReady = false; + this.pendingSeekProgress = null; + this.progressScrubbing = false; if (this.timeTotal) { this.timeTotal.textContent = ''; this.timeTotal.classList.add('hidden'); @@ -686,15 +704,33 @@ class WatchPage { } skip(seconds) { - if (this.video) { - this.video.currentTime = Math.max(0, Math.min(this.video.currentTime + seconds, this.video.duration || 0)); - } + return this.seekToContentTime(this.getCurrentPlaybackTime() + seconds); } seek(percent) { - if (this.video && this.video.duration) { - this.video.currentTime = (percent / 100) * this.video.duration; - } + return this.seekToContentTime(Number(percent) * this.getPlaybackDuration() / 100); + } + + getCurrentPlaybackTime() { + if (Number.isFinite(this.pendingSeekProgress)) return this.pendingSeekProgress; + const relative = Number(this.video?.currentTime) || 0; + return relative + (this.currentSessionId ? this.playbackTimeOffset || 0 : 0); + } + + async seekToContentTime(requestedTime) { + if (!this.video) return; + const duration = this.getPlaybackDuration(); + if (!duration || !Number.isFinite(requestedTime)) return; + const target = Math.max(0, Math.min(requestedTime, Math.max(0, duration - 0.25))); + if (!this.currentSessionId) { this.video.currentTime = target; return; } + const generation = this.playbackGeneration; + this.pendingSeekProgress = target; + this.hls?.destroy(); + this.hls = null; + this.video.pause(); + this.showLoading(); + const playlist = await this.startTranscodeSession(this.currentUrl, { ...this.currentTranscodeOptions, seekOffset: target }); + if (playlist && generation === this.playbackGeneration) this.playHls(playlist); } toggleMute() { @@ -822,17 +858,18 @@ class WatchPage { updateProgress() { if (!this.video) return; this.updateDurationDisplay(); - - const percent = (this.video.currentTime / this.video.duration) * 100; - this.progressSlider.value = percent; - this.timeCurrent.textContent = this.formatTime(this.video.currentTime); + const currentTime = this.getCurrentPlaybackTime(), duration = this.getPlaybackDuration(); + if (!this.progressScrubbing) { + this.progressSlider.value = duration > 0 ? Math.min(100, currentTime / duration * 100) : 0; + this.timeCurrent.textContent = this.formatTime(currentTime); + } // Show "Up Next" panel early for series (like streaming services do during credits) // Only show if auto-play next episode is enabled const autoPlayEnabled = this.app?.player?.settings?.autoPlayNextEpisode; if (autoPlayEnabled && this.contentType === 'series' && this.seriesInfo && !this.nextEpisodeShowing && !this.nextEpisodeDismissed) { - const duration = this.video.duration; - const currentTime = this.video.currentTime; + const duration = this.getPlaybackDuration(); + const currentTime = this.getCurrentPlaybackTime(); // Only proceed if we have reliable duration data if (isFinite(duration) && duration >= 180 && currentTime >= 120) { @@ -851,6 +888,7 @@ class WatchPage { } onMetadataLoaded() { + this.pendingSeekProgress = null; this.durationReady = true; if (!this.currentSessionId) this.setSourceDuration(this.video?.duration); else this.updateDurationDisplay(); @@ -1468,10 +1506,10 @@ class WatchPage { } async saveProgress() { - if (!this.content || !this.video || this.video.paused) return; + if (!this.content || !this.video || !this.video.src || this.video.paused) return; - const progress = Math.floor(this.video.currentTime); - const duration = Math.floor(this.video.duration); + const progress = Math.floor(this.getCurrentPlaybackTime()); + const duration = Math.floor(this.getPlaybackDuration()); if (isNaN(progress) || isNaN(duration) || duration <= 0) return; diff --git a/server/routes/remux.js b/server/routes/remux.js index 6d7031a..61553bc 100644 --- a/server/routes/remux.js +++ b/server/routes/remux.js @@ -166,7 +166,7 @@ router.get('/', async (req, res) => { }); // Cleanup on client disconnect - req.on('close', () => { + res.on('close', () => { console.log('[Remux] Client disconnected, killing FFmpeg process'); ffmpeg.kill('SIGKILL'); }); diff --git a/server/routes/subtitle.js b/server/routes/subtitle.js index 5f924ff..b884f5b 100644 --- a/server/routes/subtitle.js +++ b/server/routes/subtitle.js @@ -66,7 +66,7 @@ router.get('/', (req, res) => { // console.error(`[Subtitle FFmpeg] ${data}`); }); - req.on('close', () => { + res.on('close', () => { ffmpeg.kill('SIGKILL'); }); diff --git a/server/routes/transcode.js b/server/routes/transcode.js index 6fede23..cbc5eac 100644 --- a/server/routes/transcode.js +++ b/server/routes/transcode.js @@ -6,6 +6,8 @@ const path = require('path'); const fs = require('fs').promises; const db = require('../db'); const { requireAdmin } = require('../auth'); +const { requestLimit } = require('../services/requestLimits'); +const startupLimit = requestLimit(30); const transcodeSession = require('../services/transcodeSession'); /** @@ -30,8 +32,17 @@ transcodeSession.startCleanupInterval(); * POST /api/transcode/session * Body: { url: string, seekOffset?: number } */ -router.post('/session', async (req, res) => { - const { url, seekOffset, videoMode, videoCodec, audioCodec, audioChannels, videoHeight } = req.body; +router.post('/session', startupLimit, async (req, res) => { + const { url, seekOffset, videoMode, videoCodec, audioCodec, audioChannels, videoHeight, playbackLeaseId, requestSequence, isLive } = req.body; + const normalizedSeek = Number(seekOffset ?? 0); + if (!Number.isFinite(normalizedSeek) || normalizedSeek < 0) return res.status(400).json({ error: "Invalid seek offset" }); + let session = null, disconnected = false; + res.once("close", () => { + if (!res.writableEnded) { + disconnected = true; + if (session) transcodeSession.removeSession(session.id, req.user.id).catch(() => {}); + } + }); if (!url) { return res.status(400).json({ error: 'URL is required' }); @@ -43,11 +54,12 @@ router.post('/session', async (req, res) => { const userAgent = db.getUserAgent(settings); try { - const session = await transcodeSession.createSession(url, { + session = await transcodeSession.createSession(url, { + playbackLeaseId, requestSequence, isLive: isLive === true, ownerId: req.user.id, ffmpegPath, userAgent, - seekOffset: seekOffset || 0, + seekOffset: normalizedSeek, hwEncoder: settings.hwEncoder || 'software', maxResolution: settings.maxResolution || '1080p', quality: settings.quality || 'medium', @@ -63,11 +75,13 @@ router.post('/session', async (req, res) => { videoHeight: videoHeight // source height from probe; used to cap max-resolution in JS }); + if (disconnected) { await transcodeSession.removeSession(session.id, req.user.id); return; } await session.start(); // Wait for playlist to be ready (first segments generated) const ready = await session.waitForPlaylist(15000); + if (disconnected || session.retired) { await transcodeSession.removeSession(session.id, req.user.id); return; } if (!ready) { await transcodeSession.removeSession(session.id); return res.status(500).json({ error: 'Transcoding failed to start', reason: 'Playlist not generated in time' }); @@ -80,11 +94,26 @@ router.post('/session', async (req, res) => { }); } catch (err) { + if (session) await transcodeSession.removeSession(session.id, req.user.id); + if (disconnected || res.destroyed) return; console.error('[Transcode] Session creation failed:', redactText(err.message)); - res.status(500).json({ error: 'Failed to create session', details: err.message }); + res.status(err.statusCode || 500).json({ error: err.statusCode ? err.message : 'Failed to create session' }); } }); +router.delete('/lease', async (req, res) => { + try { + await transcodeSession.releasePlaybackLease(req.user.id, req.body.playbackLeaseId); + res.json({ success: true }); + } catch (error) { res.status(error.statusCode || 500).json({ error: error.message }); } +}); +router.post('/lease/heartbeat', (req, res) => { + try { + const found = transcodeSession.refreshPlaybackLease(req.user.id, req.body.playbackLeaseId); + res.status(found ? 200 : 404).json({ success: found }); + } catch (error) { res.status(error.statusCode || 500).json({ error: error.message }); } +}); + /** * Get HLS playlist for a session * GET /api/transcode/:sessionId/stream.m3u8 @@ -130,7 +159,7 @@ router.get('/:sessionId/:segment', async (req, res) => { } res.setHeader('Content-Type', 'video/MP2T'); - res.setHeader('Cache-Control', 'public, max-age=31536000'); // Cache forever (immutable) + res.setHeader('Cache-Control', 'private, max-age=31536000'); // Account-owned immutable segments. res.sendFile(segmentPath); }); @@ -165,7 +194,7 @@ router.get('/sessions', requireAdmin, (req, res) => { * Transcodes audio to AAC for browser compatibility while passing video through. * This fixes playback issues with Dolby/AC3/EAC3 audio that browsers can't decode. */ -router.get('/', async (req, res) => { +router.get('/', startupLimit, async (req, res) => { const { url } = req.query; try { validateHttpUrl(url); } catch (error) { return res.status(400).json({ error: error.message }); } if (!url) { @@ -258,7 +287,7 @@ router.get('/', async (req, res) => { }); // Cleanup on client disconnect - req.on('close', () => { + res.on('close', () => { console.log('[Transcode] Client disconnected, killing FFmpeg process'); ffmpeg.kill('SIGKILL'); }); diff --git a/server/services/transcodeSession.js b/server/services/transcodeSession.js index 214a56d..0d18d27 100644 --- a/server/services/transcodeSession.js +++ b/server/services/transcodeSession.js @@ -22,6 +22,24 @@ const hwDetect = require('./hwDetect'); // Session storage const sessions = new Map(); +const leases = new Map(); +let managerQueue = Promise.resolve(); +const serializeManager = task => { + const next = managerQueue.then(task); + managerQueue = next.catch(() => {}); + return next; +}; +const leaseError = (message, statusCode = 409) => Object.assign(new Error(message), { statusCode }); +function normalizePlaybackLeaseId(value) { + if (value === undefined || value === null) return null; + if (typeof value !== "string" || !/^[A-Za-z0-9_-]{16,128}$/.test(value)) throw leaseError("Invalid playback lease", 400); + return value; +} +function pruneLeases() { + const now = Date.now(); + for (const [key, lease] of leases) if (lease.until <= now) leases.delete(key); +} + // Cache directory for transcoded segments const CACHE_DIR = path.join(process.cwd(), 'transcode-cache'); @@ -29,7 +47,7 @@ const CACHE_DIR = path.join(process.cwd(), 'transcode-cache'); // Session settings const SESSION_TIMEOUT_MS = 30 * 60 * 1000; // 30 minutes idle timeout const SEGMENT_DURATION = 4; // seconds per HLS segment -const CLEANUP_INTERVAL_MS = 5 * 60 * 1000; // Check every 5 minutes +const CLEANUP_INTERVAL_MS = 30 * 1000; // Check every 5 minutes /** * Generate a unique session ID @@ -59,6 +77,13 @@ class TranscodeSession extends EventEmitter { this.id = generateSessionId(); this.url = url; this.ownerId = options.ownerId ?? null; + this.playbackLeaseId = normalizePlaybackLeaseId(options.playbackLeaseId); + this.lastHeartbeat = Date.now(); + this.retired = false; + this.stopping = false; + this.startTask = null; + this.restartTimer = null; + this.restartCount = 0; this.dir = path.join(CACHE_DIR, this.id); this.playlistPath = path.join(this.dir, 'stream.m3u8'); this.process = null; @@ -86,86 +111,75 @@ class TranscodeSession extends EventEmitter { * Start the transcoding process */ async start() { - if (this.status === 'running') { + if (this.retired) throw leaseError('Playback session was retired'); + if (this.startTask) return this.startTask; + if (this.process) return; + this.stopping = false; + const task = this.startProcess(); + this.startTask = task; + try { await task; } + catch (error) { + if (!this.retired) { this.status = 'error'; this.error = error.message; } + throw error; + } finally { if (this.startTask === task) this.startTask = null; } + } + + async startProcess() { + this.status = 'starting'; + await fs.mkdir(this.dir, { recursive: true }); + if (this.retired) throw leaseError('Playback session was retired'); + const child = spawn(this.options.ffmpegPath, this.buildFFmpegArgs(), { cwd: this.dir, windowsHide: true }); + this.process = child; + this.processStartedAt = Date.now(); + this.status = 'running'; + let processError = false; + child.stdout.on('data', () => {}); + let logBuffer = '', droppingLogLine = false; + child.stderr.on('data', data => { + logBuffer += data.toString(); + const lines = logBuffer.split('\n'); + logBuffer = lines.pop(); + for (const line of lines) { + if (droppingLogLine) droppingLogLine = false; + else console.log(`[FFmpeg ${this.id}] ${redactText(line)}`); + } + if (logBuffer.length > 65536) { logBuffer = ''; droppingLogLine = true; } + }); + child.on('error', error => { + processError = true; + this.status = 'error'; + this.error = error.message; + if (this.listenerCount('error')) this.emit('error', error); + }); + this.processClosed = new Promise(resolve => child.once('close', code => { + if (this.process === child) this.process = null; + if (!this.retired && !this.stopping && this.options.isLive && !processError) this.scheduleLiveRestart(); + else if (!processError) { + this.status = code === 0 || code === null || this.stopping ? 'stopped' : 'error'; + if (this.status === 'error') this.error = 'FFmpeg exited unsuccessfully'; + } + this.emit('exit', code); + resolve(); + })); + await this.persist(); + if (this.retired) throw leaseError('Playback session was retired'); + } + + scheduleLiveRestart() { + if (this.retired || this.stopping) return; + if (Date.now() - this.processStartedAt >= 30000) this.restartCount = 0; + if (++this.restartCount > 3) { + this.status = 'error'; + this.error = 'Live input repeatedly disconnected'; return; } - - this.status = 'starting'; - console.log(`[TranscodeSession ${this.id}] Starting session for: ${redactUrl(this.url)}`); - - // Create session directory - try { - await fs.mkdir(this.dir, { recursive: true }); - } catch (err) { - this.status = 'error'; - this.error = err.message; - throw err; - } - - // Build FFmpeg arguments for HLS output - const args = this.buildFFmpegArgs(); - - console.log(`[TranscodeSession ${this.id}] Command: ${this.options.ffmpegPath} ${redactText(args.join(' '))}`); - - try { - this.process = spawn(this.options.ffmpegPath, args, { - cwd: this.dir, - windowsHide: true - }); - - this.status = 'running'; - - // Handle stdout (should be empty for file output) - this.process.stdout.on('data', (data) => { - console.log(`[TranscodeSession ${this.id}] stdout: ${data}`); - }); - - // Handle stderr (FFmpeg progress/errors) - let stderrBuffer = ''; - this.process.stderr.on('data', (data) => { - stderrBuffer += data.toString(); - // Log periodically to avoid spam - const lines = stderrBuffer.split('\n'); - if (lines.length > 1) { - lines.slice(0, -1).forEach(line => { - if (line.trim()) { - console.log(`[FFmpeg ${this.id}] ${redactText(line)}`); - } - }); - stderrBuffer = lines[lines.length - 1]; - } - }); - - // Handle process exit - this.process.on('exit', (code) => { - if (code === 0 || code === null) { - console.log(`[TranscodeSession ${this.id}] FFmpeg completed successfully`); - this.status = 'stopped'; - } else if (code !== 255) { // 255 is often from SIGKILL - console.error(`[TranscodeSession ${this.id}] FFmpeg exited with code ${code}`); - this.status = 'error'; - this.error = `FFmpeg exited with code ${code}`; - } - this.process = null; - this.emit('exit', code); - }); - - // Handle spawn errors - this.process.on('error', (err) => { - console.error(`[TranscodeSession ${this.id}] FFmpeg error:`, redactText(err.message)); - this.status = 'error'; - this.error = err.message; - this.emit('error', err); - }); - - // Save session metadata - await this.persist(); - - } catch (err) { - this.status = 'error'; - this.error = err.message; - throw err; - } + this.status = 'reconnecting'; + this.restartTimer = setTimeout(async () => { + this.restartTimer = null; + if (this.startTask) await this.startTask.catch(() => {}); + if (!this.retired && !this.stopping) this.start().catch(error => { this.status = 'error'; this.error = error.message; }); + }, 1000); + this.restartTimer.unref?.(); } /** @@ -278,8 +292,8 @@ class TranscodeSession extends EventEmitter { args.push( '-f', 'hls', '-hls_time', String(SEGMENT_DURATION), - '-hls_list_size', '0', // Keep all segments in playlist - '-hls_flags', 'independent_segments+append_list', + '-hls_list_size', this.options.isLive ? '6' : '0', + '-hls_flags', this.options.isLive ? 'independent_segments+append_list+delete_segments+omit_endlist' : 'independent_segments+append_list', '-hls_segment_type', 'mpegts', '-hls_segment_filename', path.join(this.dir, 'seg%04d.ts'), this.playlistPath @@ -528,16 +542,16 @@ class TranscodeSession extends EventEmitter { /** * Stop the transcoding process */ - stop() { - if (this.process) { - console.log(`[TranscodeSession ${this.id}] Stopping FFmpeg process`); - this.process.kill('SIGTERM'); - // Force kill after 2 seconds if still running - setTimeout(() => { - if (this.process) { - this.process.kill('SIGKILL'); - } - }, 2000); + async stop() { + this.stopping = true; + clearTimeout(this.restartTimer); + this.restartTimer = null; + const child = this.process; + if (child) { + const force = setTimeout(() => child.kill('SIGKILL'), 2000); + force.unref?.(); + try { child.kill('SIGTERM'); await this.processClosed; } + finally { clearTimeout(force); } } this.status = 'stopped'; } @@ -569,6 +583,7 @@ class TranscodeSession extends EventEmitter { async waitForPlaylist(timeoutMs = 10000) { const startTime = Date.now(); while (Date.now() - startTime < timeoutMs) { + if (this.retired || this.status === 'error') return false; if (await this.isPlaylistReady()) { return true; } @@ -581,6 +596,7 @@ class TranscodeSession extends EventEmitter { * Get the HLS playlist content */ async getPlaylist() { + if (this.retired || this.status === 'error') return null; this.touch(); try { return await fs.readFile(this.playlistPath, 'utf8'); @@ -594,6 +610,7 @@ class TranscodeSession extends EventEmitter { */ async getSegment(segmentName) { this.touch(); + if (!/^seg[0-9]+\.ts$/.test(segmentName)) return null; const segmentPath = path.join(this.dir, segmentName); try { await fs.access(segmentPath); @@ -646,9 +663,16 @@ class TranscodeSession extends EventEmitter { * Delete session directory and all segments */ async cleanup() { - this.stop(); + this.retired = true; + clearTimeout(this.restartTimer); + if (this.startTask) await this.startTask.catch(() => {}); + await this.stop(); try { - await fs.rm(this.dir, { recursive: true, force: true }); + const target = path.resolve(this.dir); + if (path.dirname(target) !== path.resolve(CACHE_DIR) || !/^[0-9a-f]{16}$/.test(path.basename(target))) { + throw new Error('Invalid session cache directory'); + } + await fs.rm(target, { recursive: true, force: true, maxRetries: 2, retryDelay: 100 }); console.log(`[TranscodeSession ${this.id}] Cleaned up session directory`); } catch (err) { console.error(`[TranscodeSession ${this.id}] Failed to cleanup:`, err.message); @@ -665,10 +689,49 @@ class TranscodeSession extends EventEmitter { */ async function createSession(url, options = {}) { validateHttpUrl(url); - await ensureCacheDir(); - const session = new TranscodeSession(url, options); - sessions.set(session.id, session); - return session; + const playbackLeaseId = normalizePlaybackLeaseId(options.playbackLeaseId); + const key = playbackLeaseId ? `${options.ownerId}:${playbackLeaseId}` : null; + const sequence = options.requestSequence ?? 1; + if (!Number.isSafeInteger(sequence) || sequence < 1) throw leaseError('Invalid request sequence', 400); + return serializeManager(async () => { + pruneLeases(); + if (key) { + const previous = leases.get(key); + if (previous?.released || (previous && sequence <= previous.sequence)) throw leaseError('Playback request was superseded'); + if (!previous && leases.size >= 10000) throw leaseError('Too many playback requests', 429); + leases.set(key, { sequence, released: false, until: Date.now() + 5 * 60 * 1000 }); + for (const session of [...sessions.values()]) { + if (session.ownerId === options.ownerId && session.playbackLeaseId === playbackLeaseId) await removeSession(session.id, options.ownerId); + } + } + const active = [...sessions.values()].filter(session => !['stopped', 'error'].includes(session.status)); + if (active.length >= 12 || active.filter(session => session.ownerId === options.ownerId).length >= 4) throw leaseError('Too many transcode sessions', 429); + await ensureCacheDir(); + const session = new TranscodeSession(url, { ...options, playbackLeaseId }); + sessions.set(session.id, session); + return session; + }); +} + +async function releasePlaybackLease(ownerId, value) { + const leaseId = normalizePlaybackLeaseId(value); + if (!leaseId) throw leaseError('Playback lease is required', 400); + return serializeManager(async () => { + pruneLeases(); + const key = `${ownerId}:${leaseId}`; + if (!leases.has(key) && leases.size >= 10000) throw leaseError('Too many playback requests', 429); + leases.set(key, { released: true, sequence: Number.MAX_SAFE_INTEGER, until: Date.now() + 5 * 60 * 1000 }); + for (const session of [...sessions.values()]) if (session.ownerId === ownerId && session.playbackLeaseId === leaseId) await removeSession(session.id, ownerId); + }); +} + +function refreshPlaybackLease(ownerId, value) { + const leaseId = normalizePlaybackLeaseId(value); + const key = `${ownerId}:${leaseId}`, lease = leases.get(key); + if (!lease || lease.released) return false; + lease.until = Date.now() + 5 * 60 * 1000; + for (const session of sessions.values()) if (session.ownerId === ownerId && session.playbackLeaseId === leaseId) session.lastHeartbeat = Date.now(); + return true; } /** @@ -698,8 +761,8 @@ async function removeSession(sessionId, ownerId = null) { const session = sessions.get(sessionId); if (session && ownerId !== null && session.ownerId !== ownerId) return false; if (session) { - await session.cleanup(); sessions.delete(sessionId); + await session.cleanup(); return true; } return false; @@ -711,9 +774,10 @@ async function removeSession(sessionId, ownerId = null) { async function cleanupStaleSessions() { const now = Date.now(); for (const [id, session] of sessions) { - if (now - session.lastAccess > SESSION_TIMEOUT_MS) { + if ((session.playbackLeaseId && now - session.lastHeartbeat > 150000) || now - session.lastAccess > SESSION_TIMEOUT_MS) { console.log(`[TranscodeSession] Cleaning up stale session ${id}`); - await removeSession(id); + if (session.playbackLeaseId) await releasePlaybackLease(session.ownerId, session.playbackLeaseId); + else await removeSession(id); } } } @@ -772,6 +836,9 @@ function getAllSessions() { module.exports = { TranscodeSession, createSession, + releasePlaybackLease, + refreshPlaybackLease, + normalizePlaybackLeaseId, getSession, getOrCreateSession, removeSession, diff --git a/test/authentication.test.js b/test/authentication.test.js index c8dd583..3a1a0f9 100644 --- a/test/authentication.test.js +++ b/test/authentication.test.js @@ -5,6 +5,8 @@ const path = require('node:path'); const os = require('node:os'); const { spawn } = require('node:child_process'); const jwt = require('jsonwebtoken'); +const http = require('node:http'); +const { spawnSync } = require('node:child_process'); async function serverFixture(t) { const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'nodecast-auth-test-')); @@ -126,3 +128,47 @@ test('separate installations reject each other\'s cookies even when user IDs mat assert.equal((await second.request('/api/auth/me', { headers: { Cookie: one.cookie.split(';')[0] } })).status, 401); assert.equal((await first.request('/api/auth/me', { headers: { Cookie: two.cookie.split(';')[0] } })).status, 401); }); + +function ffmpegBinary() { + if (spawnSync('ffmpeg', ['-version'], { stdio: 'ignore' }).status === 0) return 'ffmpeg'; + try { const binary = require('ffmpeg-static'); if (binary && fs.existsSync(binary)) return binary; } catch {} + return null; +} +const ffmpeg = ffmpegBinary(); +test('real authenticated media requests deliver remux and owned HLS output without premature cleanup', { skip: ffmpeg ? false : 'FFmpeg unavailable' }, async t => { + const fixture = await serverFixture(t); + const admin = await fixture.request('/api/auth/setup', post({ username: 'Admin', password: 'password123' })); + const adminHeaders = { Cookie: admin.cookie.split(';')[0] }; + await fixture.request('/api/auth/users', { ...post({ username: 'Viewer', password: 'password123', role: 'viewer' }), headers: adminHeaders }); + const viewer = await fixture.request('/api/auth/login', post({ username: 'Viewer', password: 'password123' })); + const headers = { Cookie: viewer.cookie.split(';')[0] }; + const sample = path.join(fixture.dir, 'sample.ts'); + const generated = spawnSync(ffmpeg, ['-hide_banner', '-loglevel', 'error', '-y', + '-f', 'lavfi', '-i', 'testsrc2=size=64x64:rate=10', '-f', 'lavfi', '-i', 'sine=sample_rate=48000', + '-t', '12', '-c:v', 'libx264', '-threads', '1', '-preset', 'ultrafast', '-g', '20', '-c:a', 'libmp3lame', '-f', 'mpegts', sample], { encoding: 'utf8', timeout: 15000 }); + assert.equal(generated.status, 0, generated.stderr); + const bytes = fs.readFileSync(sample); + const origin = http.createServer((req, res) => { res.setHeader('Content-Length', bytes.length); res.end(bytes); }); + await new Promise(resolve => origin.listen(0, '127.0.0.1', resolve)); + t.after(() => new Promise(resolve => { origin.close(resolve); origin.closeAllConnections(); })); + const url = `http://127.0.0.1:${origin.address().port}/sample.ts`; + const remux = await fetch(`${fixture.base}/api/remux?url=${encodeURIComponent(url)}&audioCodecs=mp3`, { headers, signal: AbortSignal.timeout(10000) }); + assert.equal(remux.status, 200); + const media = Buffer.from(await remux.arrayBuffer()); + assert.ok(media.includes(Buffer.from('ftyp'))); + assert.ok(media.includes(Buffer.from('mdat'))); + const lease = 'real_fixture_playback_lease'; + const session = await fixture.request('/api/transcode/session', { ...post({ url, videoMode: 'copy', videoCodec: 'h264', playbackLeaseId: lease }), headers }); + assert.equal(session.status, 200, JSON.stringify(session.body)); + const playlist = await fetch(`${fixture.base}${session.body.playlistUrl}`, { headers }); + assert.equal(playlist.status, 200); + const text = await playlist.text(), segment = text.split(/\r?\n/).find(line => /^seg\d+\.ts$/.test(line)); + assert.ok(segment, text); + const file = await fetch(`${fixture.base}/api/transcode/${session.body.sessionId}/${segment}`, { headers }); + assert.equal(file.status, 200); + assert.match(file.headers.get('Cache-Control'), /private/); + assert.ok((await file.arrayBuffer()).byteLength > 0); + assert.equal((await fetch(`${fixture.base}${session.body.playlistUrl}`, { headers: adminHeaders })).status, 404); + assert.equal((await fixture.request('/api/transcode/lease', { ...post({ playbackLeaseId: lease }), method: 'DELETE', headers })).status, 200); + assert.equal((await fetch(`${fixture.base}${session.body.playlistUrl}`, { headers })).status, 404); +}); diff --git a/test/playback-lifecycle.test.js b/test/playback-lifecycle.test.js new file mode 100644 index 0000000..9655064 --- /dev/null +++ b/test/playback-lifecycle.test.js @@ -0,0 +1,287 @@ +const test = require('node:test'); +const assert = require('node:assert/strict'); +const fs = require('node:fs'); +const path = require('node:path'); +const os = require('node:os'); +const vm = require('node:vm'); +const { EventEmitter } = require('node:events'); +const { createRequire } = require('node:module'); +const crypto = require('node:crypto'); +const http = require('node:http'); +const { spawnSync } = require('node:child_process'); +const express = require('express'); + +function serviceFixture(t, overrides = {}, nativeSpawn = false) { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'nodecast-lifecycle-test-')); + t.after(async () => { + for (const session of module.exports.getAllSessions()) await module.exports.removeSession(session.id); + assert.equal(path.dirname(path.resolve(dir)), path.resolve(os.tmpdir())); + fs.rmSync(dir, { recursive: true, force: true }); + }); + const filename = path.join(__dirname, '../server/services/transcodeSession.js'); + const module = { exports: {} }, realRequire = createRequire(filename), children = [], restarts = []; + vm.runInNewContext(fs.readFileSync(filename, 'utf8'), { module, + process: { cwd: () => dir, env: {} }, console: { log() {}, error() {} }, + setTimeout(callback, delay) { + if (delay === 1000) { const timer = { callback, unref() {} }; restarts.push(timer); return timer; } + return setTimeout(callback, delay); + }, clearTimeout(timer) { + if (timer?.callback) { const index = restarts.indexOf(timer); if (index >= 0) restarts.splice(index, 1); } + else clearTimeout(timer); + }, setInterval, clearInterval, + require(id) { + if (id === 'child_process' && !nativeSpawn) return { spawn(binary, args) { + const child = new EventEmitter(); + child.stdout = new EventEmitter(); child.stderr = new EventEmitter(); + child.kill = signal => { child.killed = signal; queueMicrotask(() => child.emit('close', null)); }; + child.args = args; + children.push(child); + return child; + } }; + if (id === 'fs') return { promises: { ...fs.promises, ...overrides } }; + return realRequire(id); + } + }, { filename }); + return { service: module.exports, children, restarts, dir }; +} + +test('leases replace sessions atomically, reject stale requests and keep accounts separate', async t => { + const { service } = serviceFixture(t); + const options = { ownerId: 1, playbackLeaseId: 'opaque_playback_lease_1' }; + const [first, latest] = await Promise.all([1, 2].map(requestSequence => service.createSession('http://provider.test/live', { ...options, requestSequence }))); + assert.equal(first.retired, true); + assert.equal(service.getSession(first.id, 1), undefined); + assert.equal(service.getSession(latest.id, 1), latest); + await assert.rejects(service.createSession('http://provider.test/live', { ...options, requestSequence: 1 }), /superseded/); + const other = await service.createSession('http://provider.test/live', { ...options, ownerId: 2 }); + await service.releasePlaybackLease(1, options.playbackLeaseId); + assert.equal(latest.retired, true); + assert.equal(service.getSession(other.id, 2), other); + await assert.rejects(service.createSession('http://provider.test/live', { ...options, requestSequence: 100 }), /superseded/); + await service.releasePlaybackLease(1, 'released_before_arrival'); + await assert.rejects(service.createSession('http://provider.test/live', { ownerId: 1, playbackLeaseId: 'released_before_arrival' }), /superseded/); + assert.equal(service.refreshPlaybackLease(1, options.playbackLeaseId), false); + other.lastHeartbeat = Date.now() - 151000; + await service.cleanupStaleSessions(); + assert.equal(other.retired, true); + await assert.rejects(service.createSession('http://provider.test/live', { ...options, ownerId: 2, requestSequence: 2 }), /superseded/); +}); + +test('disconnecting during session startup releases the created session', async t => { + const filename = path.join(__dirname, '../server/routes/transcode.js'); + const module = { exports: {} }, realRequire = createRequire(filename); + let ready, starting; + const waiting = new Promise(resolve => { starting = resolve; }); + const gate = new Promise(resolve => { ready = resolve; }); + const removed = []; + vm.runInNewContext(fs.readFileSync(filename, 'utf8'), { module, console: { error() {} }, require(id) { + if (id === '../db') return { settings: { get: async () => ({}) }, getUserAgent: () => 'test' }; + if (id === '../auth') return { requireAdmin: (req, res, next) => next() }; + if (id === '../services/transcodeSession') return { startCleanupInterval() {}, + async createSession() { return { id: 'owned-session', start: async () => {}, waitForPlaylist() { starting(); return gate; } }; }, + async removeSession(id, owner) { removed.push({ id, owner }); } + }; + return realRequire(id); + } }, { filename }); + const app = express(); app.use(express.json()); + app.use((req, res, next) => { req.user = { id: 7 }; next(); }); + app.use('/api/transcode', module.exports); + const server = http.createServer(app); + await new Promise(resolve => server.listen(0, '127.0.0.1', resolve)); + t.after(() => new Promise(resolve => { server.close(resolve); server.closeAllConnections(); })); + const controller = new AbortController(); + const response = fetch(`http://127.0.0.1:${server.address().port}/api/transcode/session`, { + method: 'POST', signal: controller.signal, headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ url: 'http://provider.test/live' }) + }); + const rejected = assert.rejects(response, { name: 'AbortError' }); + await waiting; + controller.abort(); + await rejected; + for (let i = 0; i < 100 && !removed.length; i++) await new Promise(resolve => setTimeout(resolve, 5)); + assert.deepEqual(removed[0], { id: 'owned-session', owner: 7 }); + ready(false); +}); + +function ffmpegBinary() { + if (spawnSync('ffmpeg', ['-version'], { stdio: 'ignore' }).status === 0) return 'ffmpeg'; + try { const binary = require('ffmpeg-static'); if (binary && fs.existsSync(binary)) return binary; } catch {} + return null; +} +const ffmpeg = ffmpegBinary(); +test('real FFmpeg creates playable HLS segments and cleans its session cache', { skip: ffmpeg ? false : 'FFmpeg unavailable' }, async t => { + process.env.ALLOW_LOCAL_MEDIA_URLS = 'true'; // Synthetic loopback media origin only. + const { service, dir } = serviceFixture(t, {}, true); + const sample = path.join(dir, 'sample.ts'); + const generated = spawnSync(ffmpeg, ['-hide_banner', '-loglevel', 'error', '-y', + '-f', 'lavfi', '-i', 'testsrc2=size=64x64:rate=10', '-f', 'lavfi', '-i', 'sine=sample_rate=48000', + '-t', '12', '-c:v', 'libx264', '-threads', '1', '-preset', 'ultrafast', '-g', '20', '-c:a', 'aac', '-f', 'mpegts', sample], { encoding: 'utf8', timeout: 15000 }); + assert.equal(generated.status, 0, generated.stderr); + const bytes = fs.readFileSync(sample); + const server = http.createServer((req, res) => { res.setHeader('Content-Length', bytes.length); res.end(bytes); }); + await new Promise(resolve => server.listen(0, '127.0.0.1', resolve)); + t.after(() => new Promise(resolve => { server.close(resolve); server.closeAllConnections(); })); + const session = await service.createSession(`http://127.0.0.1:${server.address().port}/sample.ts`, + { ownerId: 1, ffmpegPath: ffmpeg, videoMode: 'copy', videoCodec: 'h264', audioCodec: 'aac', audioChannels: 2 }); + t.after(() => service.removeSession(session.id, 1)); + await session.start(); + assert.equal(await session.waitForPlaylist(5000), true, session.error || 'No playlist generated'); + const playlist = await session.getPlaylist(); + const segment = playlist.split(/\r?\n/).find(line => /^seg\d+\.ts$/.test(line)); + assert.ok(segment, playlist); + const segmentPath = await session.getSegment(segment); + const decoded = spawnSync(ffmpeg, ['-hide_banner', '-loglevel', 'error', '-i', segmentPath, '-f', 'null', '-'], { encoding: 'utf8', timeout: 15000 }); + assert.equal(decoded.status, 0, decoded.stderr); + assert.equal(await session.getSegment('../session.json'), null); + await service.removeSession(session.id, 1); + assert.equal(fs.existsSync(session.dir), false); +}); + +test('concurrent session creation respects account limits without poisoning the manager', async t => { + const { service } = serviceFixture(t); + const attempts = await Promise.allSettled(Array.from({ length: 8 }, () => service.createSession('http://provider.test/live', { ownerId: 1 }))); + const accepted = attempts.filter(result => result.status === 'fulfilled'); + assert.equal(accepted.length, 4); + assert.ok(attempts.filter(result => result.status === 'rejected').every(result => result.reason.statusCode === 429)); + for (const result of accepted) await service.removeSession(result.value.id, 1); + const recovered = await service.createSession('http://provider.test/live', { ownerId: 1 }); + await service.removeSession(recovered.id, 1); +}); + +test('retiring a session during asynchronous startup never spawns FFmpeg or leaves cache files', async t => { + let release, entered; + const blocked = new Promise(resolve => { entered = resolve; }); + const gate = new Promise(resolve => { release = resolve; }); + const { service, children } = serviceFixture(t, { async mkdir(dir, options) { + if (/transcode-cache[/\\][0-9a-f]+$/.test(dir)) { entered(); await gate; } + return fs.promises.mkdir(dir, options); + } }); + const session = await service.createSession('http://provider.test/live', { ownerId: 1 }); + const starting = assert.rejects(session.start(), /retired/); + await blocked; + const removing = service.removeSession(session.id, 1); + release(); + await Promise.all([starting, removing]); + assert.equal(children.length, 0); + assert.equal(fs.existsSync(session.dir), false); +}); + +test('cleanup waits for process close and live retries are bounded and cancellable', async t => { + const { service, children, restarts } = serviceFixture(t); + const session = await service.createSession('http://provider.test/live', { ownerId: 1, isLive: true }); + await session.start(); + assert.ok(children[0].args.includes('independent_segments+append_list+delete_segments+omit_endlist')); + for (let i = 0; i < 3; i++) { + children.at(-1).emit('close', 0); + assert.equal(session.status, 'reconnecting'); + await restarts.shift().callback(); + await session.startTask; + } + assert.equal(children.length, 4); + children.at(-1).emit('close', 1); + assert.equal(session.status, 'error'); + assert.equal(restarts.length, 0); + await service.removeSession(session.id, 1); + const active = await service.createSession('http://provider.test/live', { ownerId: 1, isLive: true }); + await active.start(); + const child = children.at(-1); + await service.removeSession(active.id, 1); + assert.equal(child.killed, 'SIGTERM'); + assert.equal(active.process, null); + assert.equal(fs.existsSync(active.dir), false); + assert.equal(restarts.length, 0); + await assert.rejects(active.start(), /retired/); +}); + +test('a subprocess spawn error becomes a failed session without an unhandled error event', async t => { + const { service, children } = serviceFixture(t); + const session = await service.createSession('http://provider.test/live', { ownerId: 1 }); + await session.start(); + children[0].emit('error', new Error('FFmpeg unavailable')); + children[0].emit('close', -1); + assert.equal(await session.waitForPlaylist(10), false); + assert.equal(session.status, 'error'); + await service.removeSession(session.id, 1); +}); + +test('browser release and replacement discard late session responses, including insecure HTTP UUID fallback', async () => { + const requests = [], pending = [], intervals = new Set(); + const window = { crypto: { getRandomValues: values => crypto.webcrypto.getRandomValues(values) }, addEventListener() {} }; + vm.runInNewContext(fs.readFileSync(path.join(__dirname, '../public/js/components/PlaybackSessions.js'), 'utf8'), { + window, AbortController, Uint8Array, setInterval: callback => { intervals.add(callback); return callback; }, clearInterval: callback => intervals.delete(callback), + fetch(url, options) { + requests.push({ url, options }); + if (url === '/api/transcode/session') return new Promise(resolve => pending.push(resolve)); + return Promise.resolve({ ok: true }); + } + }); + const playback = new window.PlaybackSessions(true); + const first = assert.rejects(playback.create('http://provider.test/one', {}), /superseded/); + const lease = JSON.parse(requests[0].options.body).playbackLeaseId; + assert.match(lease, /^[0-9a-f]{32}$/); + const active = playback.create('http://provider.test/two', {}); + pending[1]({ ok: true, json: async () => ({ sessionId: 'new', playlistUrl: '/new.m3u8' }) }); + assert.equal((await active).sessionId, 'new'); + pending[0]({ ok: true, json: async () => ({ sessionId: 'old', playlistUrl: '/old.m3u8' }) }); + await first; + assert.ok(requests.some(request => request.url === '/api/transcode/old' && request.options.method === 'DELETE')); + assert.equal(intervals.size, 1); + const stopped = assert.rejects(playback.create('http://provider.test/three', {}), /superseded/); + playback.stop(); + pending[2]({ ok: true, json: async () => ({ sessionId: 'late' }) }); + await stopped; + assert.equal(intervals.size, 0); + assert.ok(requests.some(request => request.url === '/api/transcode/lease' && JSON.parse(request.options.body).playbackLeaseId === lease)); + assert.ok(requests.some(request => request.url === '/api/transcode/late')); +}); + +test('Watch resume and session seeking preserve original content time and total duration', async () => { + const window = {}, calls = []; + const PlaybackSessions = class { async create(url, options) { calls.push({ url, options }); return { sessionId: 'owned', playlistUrl: '/owned.m3u8' }; } }; + vm.runInNewContext(fs.readFileSync(path.join(__dirname, '../public/js/pages/WatchPage.js'), 'utf8'), { + window, PlaybackSessions, VodDuration: require('../public/js/components/VodDuration') + }); + const watch = Object.create(window.WatchPage.prototype); + Object.assign(watch, { playbackGeneration: 1, resumeTime: 120, sourceDuration: 600, currentUrl: 'http://provider.test/movie', + video: { currentTime: 3, duration: 24, pause() {} }, showLoading() {}, playHls(url) { this.playlist = url; } }); + await watch.startTranscodeSession(watch.currentUrl, { videoMode: 'copy' }); + assert.equal(watch.resumeTime, 0); + assert.equal(watch.getCurrentPlaybackTime(), 123); + assert.equal(watch.getPlaybackDuration(), 600); + await watch.seek(50); + assert.equal(calls[1].options.seekOffset, 300); + assert.equal(calls[1].url, watch.currentUrl); + assert.equal(watch.getCurrentPlaybackTime(), 300); + watch.video.currentTime = 0; + watch.onMetadataLoaded(); + assert.equal(watch.getCurrentPlaybackTime(), 300); + assert.equal(watch.playlist, '/owned.m3u8'); +}); + +for (const name of ['VideoPlayer', 'WatchPage']) { + test(`${name} ignores callbacks from an obsolete HLS instance`, () => { + const window = {}; + class Hls extends EventEmitter { + static Events = { MANIFEST_PARSED: 'manifest', ERROR: 'error', SUBTITLE_TRACKS_UPDATED: 'subtitles', SUBTITLE_TRACK_SWITCH: 'switch' }; + static ErrorTypes = { NETWORK_ERROR: 'network', MEDIA_ERROR: 'media' }; + loadSource() {} attachMedia() {} destroy() { this.destroyed = true; } + } + const file = name === 'VideoPlayer' ? 'components/VideoPlayer.js' : 'pages/WatchPage.js'; + vm.runInNewContext(fs.readFileSync(path.join(__dirname, '../public/js', file), 'utf8'), { window, Hls, console: { log() {}, error() {} } }); + const player = Object.create(window[name].prototype); + let plays = 0; + player.video = { play: async () => { plays++; } }; + player.playbackGeneration = 1; + player.playHls('/old'); + const stale = player.hls, onManifest = stale.listeners('manifest')[0], onError = stale.listeners('error')[0]; + player.playbackGeneration++; + player.playHls('/new'); + const current = player.hls; + onManifest(); + onError('error', { fatal: true, type: 'network' }); + assert.equal(plays, 0); + assert.equal(current.destroyed, undefined); + current.emit('manifest'); + assert.equal(plays, 1); + }); +} diff --git a/test/remux-playback.test.js b/test/remux-playback.test.js index 9d461c1..3727d30 100644 --- a/test/remux-playback.test.js +++ b/test/remux-playback.test.js @@ -70,4 +70,18 @@ for (const name of ['VideoPlayer', 'WatchPage']) { await playing; assert.equal(player.video.src, ''); }); + test(`${name} ignores a failed automatic probe after playback stops`, async () => { + let fail, started; + const probing = new Promise(resolve => { started = resolve; }); + const player = playerFixture(name, { autoTranscode: true }, () => { + started(); + return new Promise((resolve, reject) => { fail = reject; }); + }); + const playing = player.start('http://provider.test/movie.mp4'); + await probing; + player.stop(); + fail(new Error('Provider unavailable')); + await playing; + assert.equal(player.video.src, ''); + }); }