Lock files to generate torrents/move files

server/lib/job-queue/handlers/manage-video-torrent.ts

@@ -1,5 +1,7 @@
 import { Job } from 'bullmq'
+import { extractVideo } from '@server/helpers/video'
 import { createTorrentAndSetInfoHash, updateTorrentMetadata } from '@server/helpers/webtorrent'
+import { VideoPathManager } from '@server/lib/video-path-manager'
 import { VideoModel } from '@server/models/video/video'
 import { VideoFileModel } from '@server/models/video/video-file'
 import { VideoStreamingPlaylistModel } from '@server/models/video/video-streaming-playlist'
@@ -30,6 +32,9 @@ async function doCreateAction (payload: ManageVideoTorrentPayload & { action: 'c
 
   if (!video || !file) return
 
+  const fileMutexReleaser = await VideoPathManager.Instance.lockFiles(video.uuid)
+
+  try {
     await createTorrentAndSetInfoHash(video, file)
 
     // Refresh videoFile because the createTorrentAndSetInfoHash could be long
@@ -40,7 +45,10 @@ async function doCreateAction (payload: ManageVideoTorrentPayload & { action: 'c
     refreshedFile.infoHash = file.infoHash
     refreshedFile.torrentFilename = file.torrentFilename
 
-  return refreshedFile.save()
+    await refreshedFile.save()
+  } finally {
+    fileMutexReleaser()
+  }
 }
 
 async function doUpdateMetadataAction (payload: ManageVideoTorrentPayload & { action: 'update-metadata' }) {
@@ -52,9 +60,16 @@ async function doUpdateMetadataAction (payload: ManageVideoTorrentPayload & { ac
 
   if ((!video && !streamingPlaylist) || !file) return
 
+  const extractedVideo = extractVideo(video || streamingPlaylist)
+  const fileMutexReleaser = await VideoPathManager.Instance.lockFiles(extractedVideo.uuid)
+
+  try {
     await updateTorrentMetadata(video || streamingPlaylist, file)
 
     await file.save()
+  } finally {
+    fileMutexReleaser()
+  }
 }
 
 async function loadVideoOrLog (videoId: number) {

server/lib/job-queue/handlers/move-to-object-storage.ts

@@ -28,6 +28,8 @@ export async function processMoveToObjectStorage (job: Job) {
 
   const lTags = lTagsBase(video.uuid, video.url)
 
+  const fileMutexReleaser = await VideoPathManager.Instance.lockFiles(video.uuid)
+
   try {
     if (video.VideoFiles) {
       logger.debug('Moving %d webtorrent files for video %s.', video.VideoFiles.length, video.uuid, lTags)
@@ -49,6 +51,10 @@ export async function processMoveToObjectStorage (job: Job) {
     }
   } catch (err) {
     await onMoveToObjectStorageFailure(job, err)
+
+    throw err
+  } finally {
+    fileMutexReleaser()
   }
 
   return payload.videoUUID