From b34ee7fa5f6558bd6fb870756ace1cd12e40e94c Mon Sep 17 00:00:00 2001 From: Chocobozzz Date: Wed, 25 May 2022 11:08:12 +0200 Subject: Cleanup muxing session method options --- server/lib/live/shared/muxing-session.ts | 41 ++++++++++++++++---------------- 1 file changed, 20 insertions(+), 21 deletions(-) (limited to 'server/lib') diff --git a/server/lib/live/shared/muxing-session.ts b/server/lib/live/shared/muxing-session.ts index 1ee9b430f..98a7b2613 100644 --- a/server/lib/live/shared/muxing-session.ts +++ b/server/lib/live/shared/muxing-session.ts @@ -150,8 +150,8 @@ class MuxingSession extends EventEmitter { logger.info('Running live muxing/transcoding for %s.', this.videoUUID, this.lTags()) - this.watchTSFiles(this.outDirectory) - this.watchMasterFile(this.outDirectory) + this.watchTSFiles() + this.watchMasterFile() let ffmpegShellCommand: string this.ffmpegCommand.on('start', cmdline => { @@ -161,13 +161,13 @@ class MuxingSession extends EventEmitter { }) this.ffmpegCommand.on('error', (err, stdout, stderr) => { - this.onFFmpegError({ err, stdout, stderr, outPath: this.outDirectory, ffmpegShellCommand }) + this.onFFmpegError({ err, stdout, stderr, ffmpegShellCommand }) }) this.ffmpegCommand.on('end', () => { this.emit('ffmpeg-end', ({ videoId: this.videoId })) - this.onFFmpegEnded(this.outDirectory) + this.onFFmpegEnded() }) this.ffmpegCommand.run() @@ -189,12 +189,11 @@ class MuxingSession extends EventEmitter { err: any stdout: string stderr: string - outPath: string ffmpegShellCommand: string }) { - const { err, stdout, stderr, outPath, ffmpegShellCommand } = options + const { err, stdout, stderr, ffmpegShellCommand } = options - this.onFFmpegEnded(outPath) + this.onFFmpegEnded() // Don't care that we killed the ffmpeg process if (err?.message?.includes('Exiting normally')) return @@ -204,7 +203,7 @@ class MuxingSession extends EventEmitter { this.emit('ffmpeg-error', ({ videoId: this.videoId })) } - private onFFmpegEnded (outPath: string) { + private onFFmpegEnded () { logger.info('RTMP transmuxing for video %s ended. Scheduling cleanup', this.inputUrl, this.lTags()) setTimeout(() => { @@ -214,12 +213,12 @@ class MuxingSession extends EventEmitter { .then(() => { // Process remaining segments hash for (const key of Object.keys(this.segmentsToProcessPerPlaylist)) { - this.processSegments(outPath, this.segmentsToProcessPerPlaylist[key]) + this.processSegments(this.segmentsToProcessPerPlaylist[key]) } }) .catch(err => { logger.error( - 'Cannot close watchers of %s or process remaining hash segments.', outPath, + 'Cannot close watchers of %s or process remaining hash segments.', this.outDirectory, { err, ...this.lTags() } ) }) @@ -228,21 +227,21 @@ class MuxingSession extends EventEmitter { }, 1000) } - private watchMasterFile (outPath: string) { - this.masterWatcher = watch(outPath + '/' + this.streamingPlaylist.playlistFilename) + private watchMasterFile () { + this.masterWatcher = watch(this.outDirectory + '/' + this.streamingPlaylist.playlistFilename) this.masterWatcher.on('add', () => { this.emit('master-playlist-created', { videoId: this.videoId }) this.masterWatcher.close() - .catch(err => logger.error('Cannot close master watcher of %s.', outPath, { err, ...this.lTags() })) + .catch(err => logger.error('Cannot close master watcher of %s.', this.outDirectory, { err, ...this.lTags() })) }) } - private watchTSFiles (outPath: string) { + private watchTSFiles () { const startStreamDateTime = new Date().getTime() - this.tsWatcher = watch(outPath + '/*.ts') + this.tsWatcher = watch(this.outDirectory + '/*.ts') const playlistIdMatcher = /^([\d+])-/ @@ -252,7 +251,7 @@ class MuxingSession extends EventEmitter { const playlistId = basename(segmentPath).match(playlistIdMatcher)[0] const segmentsToProcess = this.segmentsToProcessPerPlaylist[playlistId] || [] - this.processSegments(outPath, segmentsToProcess) + this.processSegments(segmentsToProcess) this.segmentsToProcessPerPlaylist[playlistId] = [ segmentPath ] @@ -273,7 +272,7 @@ class MuxingSession extends EventEmitter { } } - const deleteHandler = segmentPath => LiveSegmentShaStore.Instance.removeSegmentSha(this.videoUUID, segmentPath) + const deleteHandler = (segmentPath: string) => LiveSegmentShaStore.Instance.removeSegmentSha(this.videoUUID, segmentPath) this.tsWatcher.on('add', p => addHandler(p)) this.tsWatcher.on('unlink', p => deleteHandler(p)) @@ -332,15 +331,15 @@ class MuxingSession extends EventEmitter { return now <= max } - private processSegments (hlsVideoPath: string, segmentPaths: string[]) { + private processSegments (segmentPaths: string[]) { mapSeries(segmentPaths, async previousSegment => { // Add sha hash of previous segments, because ffmpeg should have finished generating them await LiveSegmentShaStore.Instance.addSegmentSha(this.videoUUID, previousSegment) if (this.saveReplay) { - await this.addSegmentToReplay(hlsVideoPath, previousSegment) + await this.addSegmentToReplay(previousSegment) } - }).catch(err => logger.error('Cannot process segments in %s', hlsVideoPath, { err, ...this.lTags() })) + }).catch(err => logger.error('Cannot process segments', { err, ...this.lTags() })) } private hasClientSocketInBadHealth (sessionId: string) { @@ -367,7 +366,7 @@ class MuxingSession extends EventEmitter { return false } - private async addSegmentToReplay (hlsVideoPath: string, segmentPath: string) { + private async addSegmentToReplay (segmentPath: string) { const segmentName = basename(segmentPath) const dest = join(this.replayDirectory, buildConcatenatedName(segmentName)) -- cgit v1.2.3 From 5333788c08ab6152303829d4624774b5d788ff40 Mon Sep 17 00:00:00 2001 From: Chocobozzz Date: Wed, 25 May 2022 14:54:16 +0200 Subject: Fix saving permanent live replay on quick restream --- server/lib/job-queue/handlers/video-live-ending.ts | 51 +++++++++------------- server/lib/live/live-manager.ts | 6 ++- server/lib/live/live-utils.ts | 40 +++++++++++++++-- 3 files changed, 60 insertions(+), 37 deletions(-) (limited to 'server/lib') diff --git a/server/lib/job-queue/handlers/video-live-ending.ts b/server/lib/job-queue/handlers/video-live-ending.ts index 55fd09344..79aa547ba 100644 --- a/server/lib/job-queue/handlers/video-live-ending.ts +++ b/server/lib/job-queue/handlers/video-live-ending.ts @@ -1,10 +1,10 @@ import { Job } from 'bull' -import { pathExists, readdir, remove } from 'fs-extra' +import { readdir, remove } from 'fs-extra' import { join } from 'path' import { ffprobePromise, getAudioStream, getVideoStreamDimensionsInfo, getVideoStreamDuration } from '@server/helpers/ffmpeg' import { getLocalVideoActivityPubUrl } from '@server/lib/activitypub/url' import { federateVideoIfNeeded } from '@server/lib/activitypub/videos' -import { cleanupLive, LiveSegmentShaStore } from '@server/lib/live' +import { cleanupNormalLive, cleanupPermanentLive, cleanupTMPLiveFiles, LiveSegmentShaStore } from '@server/lib/live' import { generateHLSMasterPlaylistFilename, generateHlsSha256SegmentsFilename, @@ -45,13 +45,13 @@ async function processVideoLiveEnding (job: Job) { LiveSegmentShaStore.Instance.cleanupShaSegments(liveVideo.uuid) if (live.saveReplay !== true) { - return cleanupLiveAndFederate({ liveVideo }) + return cleanupLiveAndFederate({ live, video: liveVideo }) } if (live.permanentLive) { await saveReplayToExternalVideo({ liveVideo, liveSession, publishedAt: payload.publishedAt, replayDirectory: payload.replayDirectory }) - return cleanupLiveAndFederate({ liveVideo }) + return cleanupLiveAndFederate({ live, video: liveVideo }) } return replaceLiveByReplay({ liveVideo, live, liveSession, replayDirectory: payload.replayDirectory }) @@ -164,7 +164,11 @@ async function replaceLiveByReplay (options: { await assignReplayFilesToVideo({ video: videoWithFiles, replayDirectory }) - await remove(getLiveReplayBaseDirectory(videoWithFiles)) + if (live.permanentLive) { // Remove session replay + await remove(replayDirectory) + } else { // We won't stream again in this live, we can delete the base replay directory + await remove(getLiveReplayBaseDirectory(videoWithFiles)) + } // Regenerate the thumbnail & preview? if (videoWithFiles.getMiniature().automaticallyGenerated === true) { @@ -227,34 +231,19 @@ async function assignReplayFilesToVideo (options: { } async function cleanupLiveAndFederate (options: { - liveVideo: MVideo + live: MVideoLive + video: MVideo }) { - const { liveVideo } = options - - const streamingPlaylist = await VideoStreamingPlaylistModel.loadHLSPlaylistByVideo(liveVideo.id) - await cleanupLive(liveVideo, streamingPlaylist) - - const fullVideo = await VideoModel.loadAndPopulateAccountAndServerAndTags(liveVideo.id) - return federateVideoIfNeeded(fullVideo, false, undefined) -} - -async function cleanupTMPLiveFiles (hlsDirectory: string) { - if (!await pathExists(hlsDirectory)) return + const { live, video } = options - const files = await readdir(hlsDirectory) + const streamingPlaylist = await VideoStreamingPlaylistModel.loadHLSPlaylistByVideo(video.id) - for (const filename of files) { - if ( - filename.endsWith('.ts') || - filename.endsWith('.m3u8') || - filename.endsWith('.mpd') || - filename.endsWith('.m4s') || - filename.endsWith('.tmp') - ) { - const p = join(hlsDirectory, filename) - - remove(p) - .catch(err => logger.error('Cannot remove %s.', p, { err })) - } + if (live.permanentLive) { + await cleanupPermanentLive(video, streamingPlaylist) + } else { + await cleanupNormalLive(video, streamingPlaylist) } + + const fullVideo = await VideoModel.loadAndPopulateAccountAndServerAndTags(video.id) + return federateVideoIfNeeded(fullVideo, false, undefined) } diff --git a/server/lib/live/live-manager.ts b/server/lib/live/live-manager.ts index e04ae9fef..0f14a6851 100644 --- a/server/lib/live/live-manager.ts +++ b/server/lib/live/live-manager.ts @@ -28,7 +28,7 @@ import { generateHLSMasterPlaylistFilename, generateHlsSha256SegmentsFilename, g import { PeerTubeSocket } from '../peertube-socket' import { LiveQuotaStore } from './live-quota-store' import { LiveSegmentShaStore } from './live-segment-sha-store' -import { cleanupLive } from './live-utils' +import { cleanupPermanentLive } from './live-utils' import { MuxingSession } from './shared' const NodeRtmpSession = require('node-media-server/src/node_rtmp_session') @@ -224,7 +224,9 @@ class LiveManager { const oldStreamingPlaylist = await VideoStreamingPlaylistModel.loadHLSPlaylistByVideo(video.id) if (oldStreamingPlaylist) { - await cleanupLive(video, oldStreamingPlaylist) + if (!videoLive.permanentLive) throw new Error('Found previous session in a non permanent live: ' + video.uuid) + + await cleanupPermanentLive(video, oldStreamingPlaylist) } this.videoSessions.set(video.id, sessionId) diff --git a/server/lib/live/live-utils.ts b/server/lib/live/live-utils.ts index 46c7fd2f8..6365e23db 100644 --- a/server/lib/live/live-utils.ts +++ b/server/lib/live/live-utils.ts @@ -1,5 +1,6 @@ -import { remove } from 'fs-extra' -import { basename } from 'path' +import { pathExists, readdir, remove } from 'fs-extra' +import { basename, join } from 'path' +import { logger } from '@server/helpers/logger' import { MStreamingPlaylist, MVideo } from '@server/types/models' import { getLiveDirectory } from '../paths' @@ -9,7 +10,15 @@ function buildConcatenatedName (segmentOrPlaylistPath: string) { return 'concat-' + num[1] + '.ts' } -async function cleanupLive (video: MVideo, streamingPlaylist?: MStreamingPlaylist) { +async function cleanupPermanentLive (video: MVideo, streamingPlaylist?: MStreamingPlaylist) { + const hlsDirectory = getLiveDirectory(video) + + await cleanupTMPLiveFiles(hlsDirectory) + + if (streamingPlaylist) await streamingPlaylist.destroy() +} + +async function cleanupNormalLive (video: MVideo, streamingPlaylist?: MStreamingPlaylist) { const hlsDirectory = getLiveDirectory(video) await remove(hlsDirectory) @@ -17,7 +26,30 @@ async function cleanupLive (video: MVideo, streamingPlaylist?: MStreamingPlaylis if (streamingPlaylist) await streamingPlaylist.destroy() } +async function cleanupTMPLiveFiles (hlsDirectory: string) { + if (!await pathExists(hlsDirectory)) return + + const files = await readdir(hlsDirectory) + + for (const filename of files) { + if ( + filename.endsWith('.ts') || + filename.endsWith('.m3u8') || + filename.endsWith('.mpd') || + filename.endsWith('.m4s') || + filename.endsWith('.tmp') + ) { + const p = join(hlsDirectory, filename) + + remove(p) + .catch(err => logger.error('Cannot remove %s.', p, { err })) + } + } +} + export { - cleanupLive, + cleanupPermanentLive, + cleanupNormalLive, + cleanupTMPLiveFiles, buildConcatenatedName } -- cgit v1.2.3 From c8fdfab0e36cc7324c61710009bf334e836485d9 Mon Sep 17 00:00:00 2001 From: Chocobozzz Date: Wed, 25 May 2022 15:18:29 +0200 Subject: More robust live ending job --- server/lib/job-queue/handlers/video-live-ending.ts | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) (limited to 'server/lib') diff --git a/server/lib/job-queue/handlers/video-live-ending.ts b/server/lib/job-queue/handlers/video-live-ending.ts index 79aa547ba..7607267f8 100644 --- a/server/lib/job-queue/handlers/video-live-ending.ts +++ b/server/lib/job-queue/handlers/video-live-ending.ts @@ -244,6 +244,10 @@ async function cleanupLiveAndFederate (options: { await cleanupNormalLive(video, streamingPlaylist) } - const fullVideo = await VideoModel.loadAndPopulateAccountAndServerAndTags(video.id) - return federateVideoIfNeeded(fullVideo, false, undefined) + try { + const fullVideo = await VideoModel.loadAndPopulateAccountAndServerAndTags(video.id) + return federateVideoIfNeeded(fullVideo, false, undefined) + } catch (err) { + logger.warn('Cannot federate live after cleanup', { videoId: video.id, err }) + } } -- cgit v1.2.3