Merge branch 'release/4.2.0' into develop

4 files changed, 83 insertions, 57 deletions

diff --git a/server/lib/job-queue/handlers/video-live-ending.ts b/server/lib/job-queue/handlers/video-live-ending.ts
index 55fd09344..7607267f8 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,23 @@ 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 { live, video } = options
 
-  const fullVideo = await VideoModel.loadAndPopulateAccountAndServerAndTags(liveVideo.id)
-  return federateVideoIfNeeded(fullVideo, false, undefined)
-}
+  const streamingPlaylist = await VideoStreamingPlaylistModel.loadHLSPlaylistByVideo(video.id)
 
-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)
+  if (live.permanentLive) {
+    await cleanupPermanentLive(video, streamingPlaylist)
+  } else {
+    await cleanupNormalLive(video, streamingPlaylist)
+  }
 
-      remove(p)
-        .catch(err => logger.error('Cannot remove %s.', p, { err }))
-    }
+  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 })
   }
 }
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
 }
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))