From 54d1e4398999ac75a5b75ee41dbcf6fecfb34122 Mon Sep 17 00:00:00 2001
From: ilfarpro <ilfarpro@ya.ru>
Date: Tue, 14 Oct 2025 21:15:37 +0300
Subject: [PATCH] handle deleted jobs and move on

---
 .../src/server/process/shared/common.ts       | 20 ++++-
 .../process/shared/process-storyboard.ts      | 23 ++++-
 .../server/process/shared/process-studio.ts   | 28 ++++++-
 .../process/shared/process-transcription.ts   | 28 ++++++-
 .../src/server/process/shared/process-vod.ts  | 84 +++++++++++++++++--
 5 files changed, 172 insertions(+), 11 deletions(-)

diff --git a/apps/peertube-runner/src/server/process/shared/common.ts b/apps/peertube-runner/src/server/process/shared/common.ts
index 45d6b586653..c72f4f187b0 100644
--- a/apps/peertube-runner/src/server/process/shared/common.ts
+++ b/apps/peertube-runner/src/server/process/shared/common.ts
@@ -60,14 +60,19 @@ export function scheduleTranscodingProgress (options: {
   runnerToken: string
   job: JobWithToken
   progressGetter: () => number
+  onAbort?: () => void
 }) {
-  const { job, server, progressGetter, runnerToken } = options
+  const { job, server, progressGetter, runnerToken, onAbort } = options
 
   const updateInterval = ConfigManager.Instance.isTestInstance()
     ? 500
     : 60000
 
+  let aborted = false
+
   const update = () => {
+    if (aborted) return
+
     job.progress = progressGetter() || 0
 
     server.runnerJobs.update({
@@ -75,7 +80,18 @@ export function scheduleTranscodingProgress (options: {
       jobUUID: job.uuid,
       runnerToken,
       progress: job.progress
-    }).catch(err => logger.error({ err }, 'Cannot send job progress'))
+    }).catch(err => {
+      // Job was deleted on the server, gracefully abort processing
+      if (err.res?.status === 404) {
+        logger.info({ jobUUID: job.uuid }, 'Job was deleted on the server, aborting processing')
+        aborted = true
+        clearInterval(interval)
+        if (onAbort) onAbort()
+        return
+      }
+
+      logger.error({ err }, 'Cannot send job progress')
+    })
   }
 
   const interval = setInterval(() => {
diff --git a/apps/peertube-runner/src/server/process/shared/process-storyboard.ts b/apps/peertube-runner/src/server/process/shared/process-storyboard.ts
index fdd48a972ab..5bea5ac7cab 100644
--- a/apps/peertube-runner/src/server/process/shared/process-storyboard.ts
+++ b/apps/peertube-runner/src/server/process/shared/process-storyboard.ts
@@ -13,6 +13,7 @@ export async function processGenerateStoryboard (options: ProcessOptions<RunnerJ
 
   let ffmpegProgress: number
   let videoInputPath: string
+  let jobAborted = false
 
   const outputPath = join(ConfigManager.Instance.getStoryboardDirectory(), `storyboard-${buildUUID()}.jpg`)
 
@@ -20,7 +21,10 @@ export async function processGenerateStoryboard (options: ProcessOptions<RunnerJ
     job,
     server,
     runnerToken,
-    progressGetter: () => ffmpegProgress
+    progressGetter: () => ffmpegProgress,
+    onAbort: () => {
+      jobAborted = true
+    }
   })
 
   try {
@@ -28,6 +32,11 @@ export async function processGenerateStoryboard (options: ProcessOptions<RunnerJ
 
     videoInputPath = await downloadInputFile({ url: payload.input.videoFileUrl, runnerToken, job })
 
+    if (jobAborted) {
+      logger.info(`Job ${job.uuid} was aborted, stopping processing`)
+      return
+    }
+
     logger.info(`Downloaded input file ${payload.input.videoFileUrl} for job ${job.jobToken}. Generating storyboard.`)
 
     const ffmpegImage = buildFFmpegImage()
@@ -39,6 +48,11 @@ export async function processGenerateStoryboard (options: ProcessOptions<RunnerJ
       sprites: payload.sprites
     })
 
+    if (jobAborted) {
+      logger.info(`Job ${job.uuid} was aborted during storyboard generation, stopping processing`)
+      return
+    }
+
     const successBody: GenerateStoryboardSuccess = {
       storyboardFile: outputPath
     }
@@ -50,6 +64,13 @@ export async function processGenerateStoryboard (options: ProcessOptions<RunnerJ
       payload: successBody,
       reqPayload: payload
     })
+  } catch (err) {
+    // If job was aborted, don't report the error
+    if (jobAborted) {
+      logger.info(`Job ${job.uuid} processing stopped after abort`)
+      return
+    }
+    throw err
   } finally {
     if (videoInputPath) await remove(videoInputPath)
     if (outputPath) await remove(outputPath)
diff --git a/apps/peertube-runner/src/server/process/shared/process-studio.ts b/apps/peertube-runner/src/server/process/shared/process-studio.ts
index 6a4140846e0..b1d47bd093d 100644
--- a/apps/peertube-runner/src/server/process/shared/process-studio.ts
+++ b/apps/peertube-runner/src/server/process/shared/process-studio.ts
@@ -36,12 +36,16 @@ export async function processStudioTranscoding (options: ProcessOptions<RunnerJo
   let outputPath: string
 
   let tasksProgress = 0
+  let jobAborted = false
 
   const updateProgressInterval = scheduleTranscodingProgress({
     job,
     server,
     runnerToken,
-    progressGetter: () => tasksProgress
+    progressGetter: () => tasksProgress,
+    onAbort: () => {
+      jobAborted = true
+    }
   })
 
   try {
@@ -50,12 +54,22 @@ export async function processStudioTranscoding (options: ProcessOptions<RunnerJo
     videoInputPath = await downloadInputFile({ url: payload.input.videoFileUrl, runnerToken, job })
     separatedAudioInputPath = await downloadSeparatedAudioFileIfNeeded({ urls: payload.input.separatedAudioFileUrl, runnerToken, job })
 
+    if (jobAborted) {
+      logger.info(`Job ${job.uuid} was aborted, stopping processing`)
+      return
+    }
+
     tmpVideoInputFilePath = videoInputPath
     tmpSeparatedAudioInputFilePath = separatedAudioInputPath
 
     logger.info(`Input file ${payload.input.videoFileUrl} downloaded for job ${job.jobToken}. Running studio transcoding tasks.`)
 
     for (const task of payload.tasks) {
+      if (jobAborted) {
+        logger.info(`Job ${job.uuid} was aborted during processing, stopping`)
+        return
+      }
+
       const outputFilename = 'output-edition-' + buildUUID() + '.mp4'
       outputPath = join(ConfigManager.Instance.getTranscodingDirectory(), outputFilename)
 
@@ -78,6 +92,11 @@ export async function processStudioTranscoding (options: ProcessOptions<RunnerJo
       tasksProgress += Math.floor(100 / payload.tasks.length)
     }
 
+    if (jobAborted) {
+      logger.info(`Job ${job.uuid} was aborted, stopping processing`)
+      return
+    }
+
     const successBody: VideoStudioTranscodingSuccess = {
       videoFile: outputPath
     }
@@ -89,6 +108,13 @@ export async function processStudioTranscoding (options: ProcessOptions<RunnerJo
       payload: successBody,
       reqPayload: payload
     })
+  } catch (err) {
+    // If job was aborted, don't report the error
+    if (jobAborted) {
+      logger.info(`Job ${job.uuid} processing stopped after abort`)
+      return
+    }
+    throw err
   } finally {
     if (tmpVideoInputFilePath) await remove(tmpVideoInputFilePath)
     if (tmpSeparatedAudioInputFilePath) await remove(tmpSeparatedAudioInputFilePath)
diff --git a/apps/peertube-runner/src/server/process/shared/process-transcription.ts b/apps/peertube-runner/src/server/process/shared/process-transcription.ts
index 246f1d32195..48dc53b307e 100644
--- a/apps/peertube-runner/src/server/process/shared/process-transcription.ts
+++ b/apps/peertube-runner/src/server/process/shared/process-transcription.ts
@@ -17,12 +17,16 @@ export async function processVideoTranscription (options: ProcessOptions<RunnerJ
   const payload = job.payload
 
   let inputPath: string
+  let jobAborted = false
 
   const updateProgressInterval = scheduleTranscodingProgress({
     job,
     server,
     runnerToken,
-    progressGetter: () => undefined
+    progressGetter: () => undefined,
+    onAbort: () => {
+      jobAborted = true
+    }
   })
 
   const outputPath = join(ConfigManager.Instance.getTranscriptionDirectory(), buildSUUID())
@@ -38,6 +42,11 @@ export async function processVideoTranscription (options: ProcessOptions<RunnerJ
 
     inputPath = await downloadInputFile({ url: payload.input.videoFileUrl, runnerToken, job })
 
+    if (jobAborted) {
+      logger.info(`Job ${job.uuid} was aborted, stopping processing`)
+      return
+    }
+
     logger.info(`Downloaded input file ${payload.input.videoFileUrl} for job ${job.jobToken}. Running transcription.`)
 
     if (await hasAudioStream(inputPath) !== true) {
@@ -51,6 +60,11 @@ export async function processVideoTranscription (options: ProcessOptions<RunnerJ
       return
     }
 
+    if (jobAborted) {
+      logger.info(`Job ${job.uuid} was aborted, stopping processing`)
+      return
+    }
+
     const transcriptFile = await transcriber.transcribe({
       mediaFilePath: inputPath,
       model: config.modelPath
@@ -60,6 +74,11 @@ export async function processVideoTranscription (options: ProcessOptions<RunnerJ
       transcriptDirectory: outputPath
     })
 
+    if (jobAborted) {
+      logger.info(`Job ${job.uuid} was aborted during transcription, stopping processing`)
+      return
+    }
+
     const successBody: TranscriptionSuccess = {
       inputLanguage: transcriptFile.language,
       vttFile: transcriptFile.path
@@ -72,6 +91,13 @@ export async function processVideoTranscription (options: ProcessOptions<RunnerJ
       payload: successBody,
       reqPayload: payload
     })
+  } catch (err) {
+    // If job was aborted, don't report the error
+    if (jobAborted) {
+      logger.info(`Job ${job.uuid} processing stopped after abort`)
+      return
+    }
+    throw err
   } finally {
     if (inputPath) await remove(inputPath)
     if (outputPath) await remove(outputPath)
diff --git a/apps/peertube-runner/src/server/process/shared/process-vod.ts b/apps/peertube-runner/src/server/process/shared/process-vod.ts
index d792626b779..18086842f86 100644
--- a/apps/peertube-runner/src/server/process/shared/process-vod.ts
+++ b/apps/peertube-runner/src/server/process/shared/process-vod.ts
@@ -21,6 +21,7 @@ import {
 import {
   canDoQuickAudioTranscode,
   canDoQuickVideoTranscode,
+  FFmpegVOD,
   ffprobePromise,
   getVideoStreamDimensionsInfo,
   getVideoStreamFPS
@@ -34,6 +35,8 @@ export async function processWebVideoTranscoding (options: ProcessOptions<Runner
   let ffmpegProgress: number
   let videoInputPath: string
   let separatedAudioInputPath: string
+  let ffmpegVod: FFmpegVOD
+  let jobAborted = false
 
   const outputPath = join(ConfigManager.Instance.getTranscodingDirectory(), `output-${buildUUID()}.mp4`)
 
@@ -41,7 +44,10 @@ export async function processWebVideoTranscoding (options: ProcessOptions<Runner
     job,
     server,
     runnerToken,
-    progressGetter: () => ffmpegProgress
+    progressGetter: () => ffmpegProgress,
+    onAbort: () => {
+      jobAborted = true
+    }
   })
 
   try {
@@ -50,9 +56,14 @@ export async function processWebVideoTranscoding (options: ProcessOptions<Runner
     videoInputPath = await downloadInputFile({ url: payload.input.videoFileUrl, runnerToken, job })
     separatedAudioInputPath = await downloadSeparatedAudioFileIfNeeded({ urls: payload.input.separatedAudioFileUrl, runnerToken, job })
 
+    if (jobAborted) {
+      logger.info(`Job ${job.uuid} was aborted, stopping processing`)
+      return
+    }
+
     logger.info(`Downloaded input file ${payload.input.videoFileUrl} for job ${job.jobToken}. Running web video transcoding.`)
 
-    const ffmpegVod = buildFFmpegVOD({
+    ffmpegVod = buildFFmpegVOD({
       onJobProgress: progress => {
         ffmpegProgress = progress
       }
@@ -72,6 +83,11 @@ export async function processWebVideoTranscoding (options: ProcessOptions<Runner
       fps: payload.output.fps
     })
 
+    if (jobAborted) {
+      logger.info(`Job ${job.uuid} was aborted during transcoding, stopping processing`)
+      return
+    }
+
     const successBody: VODWebVideoTranscodingSuccess = {
       videoFile: outputPath
     }
@@ -83,6 +99,13 @@ export async function processWebVideoTranscoding (options: ProcessOptions<Runner
       payload: successBody,
       reqPayload: payload
     })
+  } catch (err) {
+    // If job was aborted, don't report the error
+    if (jobAborted) {
+      logger.info(`Job ${job.uuid} processing stopped after abort`)
+      return
+    }
+    throw err
   } finally {
     if (videoInputPath) await remove(videoInputPath)
     if (separatedAudioInputPath) await remove(separatedAudioInputPath)
@@ -98,6 +121,8 @@ export async function processHLSTranscoding (options: ProcessOptions<RunnerJobVO
   let ffmpegProgress: number
   let videoInputPath: string
   let separatedAudioInputPath: string
+  let ffmpegVod: FFmpegVOD
+  let jobAborted = false
 
   const uuid = buildUUID()
   const outputPath = join(ConfigManager.Instance.getTranscodingDirectory(), `${uuid}-${payload.output.resolution}.m3u8`)
@@ -108,7 +133,10 @@ export async function processHLSTranscoding (options: ProcessOptions<RunnerJobVO
     job,
     server,
     runnerToken,
-    progressGetter: () => ffmpegProgress
+    progressGetter: () => ffmpegProgress,
+    onAbort: () => {
+      jobAborted = true
+    }
   })
 
   try {
@@ -117,6 +145,11 @@ export async function processHLSTranscoding (options: ProcessOptions<RunnerJobVO
     videoInputPath = await downloadInputFile({ url: payload.input.videoFileUrl, runnerToken, job })
     separatedAudioInputPath = await downloadSeparatedAudioFileIfNeeded({ urls: payload.input.separatedAudioFileUrl, runnerToken, job })
 
+    if (jobAborted) {
+      logger.info(`Job ${job.uuid} was aborted, stopping processing`)
+      return
+    }
+
     const inputProbe = await ffprobePromise(videoInputPath)
     const { resolution } = await getVideoStreamDimensionsInfo(videoInputPath, inputProbe)
     const fps = await getVideoStreamFPS(videoInputPath, inputProbe)
@@ -128,9 +161,14 @@ export async function processHLSTranscoding (options: ProcessOptions<RunnerJobVO
       resolution === payload.output.resolution &&
       (!resolution || fps === payload.output.fps)
 
+    if (jobAborted) {
+      logger.info(`Job ${job.uuid} was aborted, stopping processing`)
+      return
+    }
+
     logger.info(`Downloaded input file ${payload.input.videoFileUrl} for job ${job.jobToken}. Running HLS transcoding.`)
 
-    const ffmpegVod = buildFFmpegVOD({
+    ffmpegVod = buildFFmpegVOD({
       onJobProgress: progress => {
         ffmpegProgress = progress
       }
@@ -153,6 +191,11 @@ export async function processHLSTranscoding (options: ProcessOptions<RunnerJobVO
       separatedAudio: payload.output.separatedAudio
     })
 
+    if (jobAborted) {
+      logger.info(`Job ${job.uuid} was aborted during transcoding, stopping processing`)
+      return
+    }
+
     const successBody: VODHLSTranscodingSuccess = {
       resolutionPlaylistFile: outputPath,
       videoFile: videoPath
@@ -165,6 +208,13 @@ export async function processHLSTranscoding (options: ProcessOptions<RunnerJobVO
       payload: successBody,
       reqPayload: payload
     })
+  } catch (err) {
+    // If job was aborted, don't report the error
+    if (jobAborted) {
+      logger.info(`Job ${job.uuid} processing stopped after abort`)
+      return
+    }
+    throw err
   } finally {
     if (videoInputPath) await remove(videoInputPath)
     if (separatedAudioInputPath) await remove(separatedAudioInputPath)
@@ -181,6 +231,8 @@ export async function processAudioMergeTranscoding (options: ProcessOptions<Runn
   let ffmpegProgress: number
   let audioPath: string
   let previewPath: string
+  let ffmpegVod: FFmpegVOD
+  let jobAborted = false
 
   const outputPath = join(ConfigManager.Instance.getTranscodingDirectory(), `output-${buildUUID()}.mp4`)
 
@@ -188,7 +240,10 @@ export async function processAudioMergeTranscoding (options: ProcessOptions<Runn
     job,
     server,
     runnerToken,
-    progressGetter: () => ffmpegProgress
+    progressGetter: () => ffmpegProgress,
+    onAbort: () => {
+      jobAborted = true
+    }
   })
 
   try {
@@ -200,12 +255,17 @@ export async function processAudioMergeTranscoding (options: ProcessOptions<Runn
     audioPath = await downloadInputFile({ url: payload.input.audioFileUrl, runnerToken, job })
     previewPath = await downloadInputFile({ url: payload.input.previewFileUrl, runnerToken, job })
 
+    if (jobAborted) {
+      logger.info(`Job ${job.uuid} was aborted, stopping processing`)
+      return
+    }
+
     logger.info(
       `Downloaded input files ${payload.input.audioFileUrl} and ${payload.input.previewFileUrl} ` +
         `for job ${job.jobToken}. Running audio merge transcoding.`
     )
 
-    const ffmpegVod = buildFFmpegVOD({
+    ffmpegVod = buildFFmpegVOD({
       onJobProgress: progress => {
         ffmpegProgress = progress
       }
@@ -225,6 +285,11 @@ export async function processAudioMergeTranscoding (options: ProcessOptions<Runn
       fps: payload.output.fps
     })
 
+    if (jobAborted) {
+      logger.info(`Job ${job.uuid} was aborted during transcoding, stopping processing`)
+      return
+    }
+
     const successBody: VODAudioMergeTranscodingSuccess = {
       videoFile: outputPath
     }
@@ -236,6 +301,13 @@ export async function processAudioMergeTranscoding (options: ProcessOptions<Runn
       payload: successBody,
       reqPayload: payload
     })
+  } catch (err) {
+    // If job was aborted, don't report the error
+    if (jobAborted) {
+      logger.info(`Job ${job.uuid} processing stopped after abort`)
+      return
+    }
+    throw err
   } finally {
     if (audioPath) await remove(audioPath)
     if (previewPath) await remove(previewPath)