More robust sync

server/core/lib/sync-channel.ts

@@ -1,4 +1,4 @@
-import { VideoChannelSyncState } from '@peertube/peertube-models'
+import { VideoChannelSyncState, VideoImportState } from '@peertube/peertube-models'
 import { logger, loggerTagsFactory, LoggerTagsFn } from '@server/helpers/logger.js'
 import { YoutubeDlImportError, YoutubeDlImportErrorCode, YoutubeDLWrapper } from '@server/helpers/youtube-dl/index.js'
 import { CONFIG } from '@server/initializers/config.js'
@@ -50,6 +50,9 @@ export async function synchronizeChannel (options: {
     )
 
     const children: CreateJobTypeAndPayload[] = []
+    // Ids of video imports already persisted in DB
+    // If job creation fails, these must be reverted to FAILED so they are picked up by the retry mechanism instead of staying stuck
+    const touchedVideoImportIds: number[] = []
 
     let buildJobErrors = 0
 
@@ -59,7 +62,7 @@ export async function synchronizeChannel (options: {
       try {
         if (await skipImport({ channel, channelSync, targetUrl, lTags })) continue
 
-        const { job } = await buildYoutubeDLImport({
+        const { job, videoImport } = await buildYoutubeDLImport({
           user,
           channel,
           targetUrl,
@@ -72,6 +75,7 @@ export async function synchronizeChannel (options: {
         })
 
         children.push(job)
+        touchedVideoImportIds.push(videoImport.id)
       } catch (err) {
         if (err instanceof YoutubeDlImportError) {
           if (
@@ -106,6 +110,7 @@ export async function synchronizeChannel (options: {
         )
 
         children.push(await buildRetryImportJob(videoImport))
+        touchedVideoImportIds.push(videoImport.id)
       }
     }
 
@@ -118,7 +123,20 @@ export async function synchronizeChannel (options: {
       }
     }
 
-    await JobQueue.Instance.createJobWithChildren(parent, children)
+    try {
+      await JobQueue.Instance.createJobWithChildren(parent, children)
+    } catch (err) {
+      try {
+        await VideoImportModel.updateStateByIds(touchedVideoImportIds, VideoImportState.FAILED, 'Failed to create the video import job')
+      } catch (updateErr) {
+        logger.error(`Failed to update state of video imports to FAILED after failing to create the video import job`, {
+          updateErr,
+          ...rootLTags()
+        })
+      }
+
+      throw err
+    }
   } catch (err) {
     logger.error(`Failed to import ${externalChannelUrl} in channel ${channelUsername}`, { err, ...rootLTags() })
 

server/core/lib/video-pre-import.ts

@@ -239,50 +239,66 @@ export async function buildYoutubeDLImport (options: {
     videoPasswords: importDataOverride.videoPasswords
   })
 
-  await sequelizeTypescript.transaction(async transaction => {
-    // Priority to explicitly set description
-    if (importDataOverride.description) {
-      const inserted = await replaceChaptersFromDescriptionIfNeeded({ newDescription: importDataOverride.description, video, transaction })
-      if (inserted) return
+  try {
+    await sequelizeTypescript.transaction(async transaction => {
+      // Priority to explicitly set description
+      if (importDataOverride.description) {
+        const inserted = await replaceChaptersFromDescriptionIfNeeded({
+          newDescription: importDataOverride.description,
+          video,
+          transaction
+        })
+        if (inserted) return
+      }
+
+      // Then priority to youtube-dl chapters
+      if (youtubeDLInfo.chapters.length !== 0) {
+        logger.info(
+          `Inserting chapters in video ${video.uuid} from youtube-dl`,
+          { chapters: youtubeDLInfo.chapters, tags: [ 'chapters', video.uuid ] }
+        )
+
+        await replaceChapters({ video, chapters: youtubeDLInfo.chapters, transaction })
+        return
+      }
+
+      if (video.description) {
+        await replaceChaptersFromDescriptionIfNeeded({ newDescription: video.description, video, transaction })
+      }
+    })
+
+    // Get video subtitles
+    await processYoutubeSubtitles({ youtubeDL, targetUrl, video, userLanguage })
+
+    let fileExt = `.${youtubeDLInfo.ext}`
+    if (!isVideoFileExtnameValid(fileExt)) fileExt = '.mp4'
+
+    const payload: VideoImportPayload = {
+      type: 'youtube-dl' as 'youtube-dl',
+      videoImportId: videoImport.id,
+      fileExt,
+      generateTranscription: importDataOverride.generateTranscription ?? true,
+      // If part of a sync process, there is a parent job that will aggregate children results
+      preventException: !!channelSync
     }
 
-    // Then priority to youtube-dl chapters
-    if (youtubeDLInfo.chapters.length !== 0) {
-      logger.info(
-        `Inserting chapters in video ${video.uuid} from youtube-dl`,
-        { chapters: youtubeDLInfo.chapters, tags: [ 'chapters', video.uuid ] }
-      )
+    videoImport.payload = payload
+    await videoImport.save()
 
-      await replaceChapters({ video, chapters: youtubeDLInfo.chapters, transaction })
-      return
+    return {
+      videoImport,
+      job: { type: 'video-import' as 'video-import', payload }
+    }
+  } catch (err) {
+    // Auto destroy video import  to not keep a "PENDING" import that never gets a job
+    try {
+      await videoImport.Video.destroy()
+      await videoImport.destroy()
+    } catch (cleanupErr) {
+      logger.error(`Cannot cleanup video import for ${targetUrl} after a build error.`, { err: cleanupErr })
     }
 
-    if (video.description) {
-      await replaceChaptersFromDescriptionIfNeeded({ newDescription: video.description, video, transaction })
-    }
-  })
-
-  // Get video subtitles
-  await processYoutubeSubtitles({ youtubeDL, targetUrl, video, userLanguage })
-
-  let fileExt = `.${youtubeDLInfo.ext}`
-  if (!isVideoFileExtnameValid(fileExt)) fileExt = '.mp4'
-
-  const payload: VideoImportPayload = {
-    type: 'youtube-dl' as 'youtube-dl',
-    videoImportId: videoImport.id,
-    fileExt,
-    generateTranscription: importDataOverride.generateTranscription ?? true,
-    // If part of a sync process, there is a parent job that will aggregate children results
-    preventException: !!channelSync
-  }
-
-  videoImport.payload = payload
-  await videoImport.save()
-
-  return {
-    videoImport,
-    job: { type: 'video-import' as 'video-import', payload }
+    throw err
   }
 }
 

server/core/models/video/video-import.ts

@@ -237,6 +237,20 @@ export class VideoImportModel extends SequelizeModel<VideoImportModel> {
     })
   }
 
+  static updateStateByIds (ids: number[], state: VideoImportStateType, error?: string) {
+    if (ids.length === 0) return
+    return VideoImportModel.update(
+      { state, error },
+      {
+        where: {
+          id: {
+            [Op.in]: ids
+          }
+        }
+      }
+    )
+  }
+
   // ---------------------------------------------------------------------------
 
   getTargetIdentifier () {