Merge branch 'release/5.0.0' into develop

[github/Chocobozzz/PeerTube.git] / server / lib / job-queue / handlers / video-transcoding.ts

diff --git a/server/lib/job-queue/handlers/video-transcoding.ts b/server/lib/job-queue/handlers/video-transcoding.ts
index 8dbae8c423490a1679da4506544f382bf35b2c0a..3e6d2336339a7552514bb801f8c86af72aabbd32 100644 (file)
--- a/server/lib/job-queue/handlers/video-transcoding.ts
+++ b/server/lib/job-queue/handlers/video-transcoding.ts
@@ -1,7 +1,7 @@
 import { Job } from 'bullmq'
 import { TranscodeVODOptionsType } from '@server/helpers/ffmpeg'
 import { Hooks } from '@server/lib/plugins/hooks'
-import { addTranscodingJob, getTranscodingJobPriority } from '@server/lib/video'
+import { buildTranscodingJob, getTranscodingJobPriority } from '@server/lib/video'
 import { VideoPathManager } from '@server/lib/video-path-manager'
 import { moveToFailedTranscodingState, moveToNextState } from '@server/lib/video-state'
 import { UserModel } from '@server/models/user/user'
@@ -13,7 +13,6 @@ import {
   MergeAudioTranscodingPayload,
   NewWebTorrentResolutionTranscodingPayload,
   OptimizeTranscodingPayload,
-  VideoResolution,
   VideoTranscodingPayload
 } from '@shared/models'
 import { retryTransactionWrapper } from '../../../helpers/database-utils'
@@ -27,6 +26,7 @@ import {
   optimizeOriginalVideofile,
   transcodeNewWebTorrentResolution
 } from '../../transcoding/transcoding'
+import { JobQueue } from '../job-queue'
 
 type HandlerFunction = (job: Job, payload: VideoTranscodingPayload, video: MVideoFullLight, user: MUser) => Promise<void>
 
@@ -41,7 +41,7 @@ const lTags = loggerTagsFactory('transcoding')
 
 async function processVideoTranscoding (job: Job) {
   const payload = job.data as VideoTranscodingPayload
-  logger.info('Processing transcoding job %d.', job.id, lTags(payload.videoUUID))
+  logger.info('Processing transcoding job %s.', job.id, lTags(payload.videoUUID))
 
   const video = await VideoModel.loadFull(payload.videoUUID)
   // No video, maybe deleted?
@@ -93,15 +93,24 @@ async function handleHLSJob (job: Job, payload: HLSTranscodingPayload, video: MV
 
   const videoOrStreamingPlaylist = videoFileInput.getVideoOrStreamingPlaylist()
 
-  await VideoPathManager.Instance.makeAvailableVideoFile(videoFileInput.withVideoOrPlaylist(videoOrStreamingPlaylist), videoInputPath => {
-    return generateHlsPlaylistResolution({
-      video,
-      videoInputPath,
-      resolution: payload.resolution,
-      copyCodecs: payload.copyCodecs,
-      job
+  const inputFileMutexReleaser = await VideoPathManager.Instance.lockFiles(video.uuid)
+
+  try {
+    await videoFileInput.getVideo().reload()
+
+    await VideoPathManager.Instance.makeAvailableVideoFile(videoFileInput.withVideoOrPlaylist(videoOrStreamingPlaylist), videoInputPath => {
+      return generateHlsPlaylistResolution({
+        video,
+        videoInputPath,
+        inputFileMutexReleaser,
+        resolution: payload.resolution,
+        copyCodecs: payload.copyCodecs,
+        job
+      })
     })
-  })
+  } finally {
+    inputFileMutexReleaser()
+  }
 
   logger.info('HLS transcoding job for %s ended.', video.uuid, lTags(video.uuid))
 
@@ -176,38 +185,44 @@ async function onVideoFirstWebTorrentTranscoding (
   transcodeType: TranscodeVODOptionsType,
   user: MUserId
 ) {
-  const { resolution, audioStream } = await videoArg.probeMaxQualityFile()
-
-  // Maybe the video changed in database, refresh it
-  const videoDatabase = await VideoModel.loadFull(videoArg.uuid)
-  // Video does not exist anymore
-  if (!videoDatabase) return undefined
-
-  // Generate HLS version of the original file
-  const originalFileHLSPayload = {
-    ...payload,
-
-    hasAudio: !!audioStream,
-    resolution: videoDatabase.getMaxQualityFile().resolution,
-    // If we quick transcoded original file, force transcoding for HLS to avoid some weird playback issues
-    copyCodecs: transcodeType !== 'quick-transcode',
-    isMaxQuality: true
-  }
-  const hasHls = await createHlsJobIfEnabled(user, originalFileHLSPayload)
-  const hasNewResolutions = await createLowerResolutionsJobs({
-    video: videoDatabase,
-    user,
-    videoFileResolution: resolution,
-    hasAudio: !!audioStream,
-    type: 'webtorrent',
-    isNewVideo: payload.isNewVideo ?? true
-  })
-
-  await VideoJobInfoModel.decrease(videoDatabase.uuid, 'pendingTranscode')
-
-  // Move to next state if there are no other resolutions to generate
-  if (!hasHls && !hasNewResolutions) {
-    await retryTransactionWrapper(moveToNextState, { video: videoDatabase, isNewVideo: payload.isNewVideo })
+  const mutexReleaser = await VideoPathManager.Instance.lockFiles(videoArg.uuid)
+
+  try {
+    // Maybe the video changed in database, refresh it
+    const videoDatabase = await VideoModel.loadFull(videoArg.uuid)
+    // Video does not exist anymore
+    if (!videoDatabase) return undefined
+
+    const { resolution, audioStream } = await videoDatabase.probeMaxQualityFile()
+
+    // Generate HLS version of the original file
+    const originalFileHLSPayload = {
+      ...payload,
+
+      hasAudio: !!audioStream,
+      resolution: videoDatabase.getMaxQualityFile().resolution,
+      // If we quick transcoded original file, force transcoding for HLS to avoid some weird playback issues
+      copyCodecs: transcodeType !== 'quick-transcode',
+      isMaxQuality: true
+    }
+    const hasHls = await createHlsJobIfEnabled(user, originalFileHLSPayload)
+    const hasNewResolutions = await createLowerResolutionsJobs({
+      video: videoDatabase,
+      user,
+      videoFileResolution: resolution,
+      hasAudio: !!audioStream,
+      type: 'webtorrent',
+      isNewVideo: payload.isNewVideo ?? true
+    })
+
+    await VideoJobInfoModel.decrease(videoDatabase.uuid, 'pendingTranscode')
+
+    // Move to next state if there are no other resolutions to generate
+    if (!hasHls && !hasNewResolutions) {
+      await retryTransactionWrapper(moveToNextState, { video: videoDatabase, isNewVideo: payload.isNewVideo })
+    }
+  } finally {
+    mutexReleaser()
   }
 }
 
@@ -248,7 +263,7 @@ async function createHlsJobIfEnabled (user: MUserId, payload: {
     ...pick(payload, [ 'videoUUID', 'resolution', 'copyCodecs', 'isMaxQuality', 'isNewVideo', 'hasAudio' ])
   }
 
-  await addTranscodingJob(hlsTranscodingPayload, jobOptions)
+  await JobQueue.Instance.createJob(await buildTranscodingJob(hlsTranscodingPayload, jobOptions))
 
   return true
 }
@@ -265,7 +280,7 @@ async function createLowerResolutionsJobs (options: {
 
   // Create transcoding jobs if there are enabled resolutions
   const resolutionsEnabled = await Hooks.wrapObject(
-    computeResolutionsToTranscode({ input: videoFileResolution, type: 'vod', includeInput: false, strictLower: true }),
+    computeResolutionsToTranscode({ input: videoFileResolution, type: 'vod', includeInput: false, strictLower: true, hasAudio }),
     'filter:transcoding.auto.resolutions-to-transcode.result',
     options
   )
@@ -273,8 +288,6 @@ async function createLowerResolutionsJobs (options: {
   const resolutionCreated: string[] = []
 
   for (const resolution of resolutionsEnabled) {
-    if (resolution === VideoResolution.H_NOVIDEO && hasAudio === false) continue
-
     let dataInput: VideoTranscodingPayload
 
     if (CONFIG.TRANSCODING.WEBTORRENT.ENABLED && type === 'webtorrent') {
@@ -312,7 +325,7 @@ async function createLowerResolutionsJobs (options: {
       priority: await getTranscodingJobPriority(user)
     }
 
-    await addTranscodingJob(dataInput, jobOptions)
+    await JobQueue.Instance.createJob(await buildTranscodingJob(dataInput, jobOptions))
   }
 
   if (resolutionCreated.length === 0) {