Avoid aborting completing jobs

server/helpers/database-utils.ts

@@ -68,6 +68,14 @@ function transactionRetryer <T> (func: (err: any, data: T) => any) {
   })
 }
 
+function saveInTransactionWithRetries <T extends Pick<Model, 'save'>> (model: T) {
+  return retryTransactionWrapper(() => {
+    return sequelizeTypescript.transaction(async transaction => {
+      await model.save({ transaction })
+    })
+  })
+}
+
 // ---------------------------------------------------------------------------
 
 function resetSequelizeInstance <T> (instance: Model<T>) {
@@ -105,6 +113,7 @@ export {
   resetSequelizeInstance,
   retryTransactionWrapper,
   transactionRetryer,
+  saveInTransactionWithRetries,
   afterCommitIfTransaction,
   filterNonExistingModels,
   deleteAllModels,

server/initializers/constants.ts

@@ -577,6 +577,7 @@ const VIDEO_PLAYLIST_TYPES: { [ id in VideoPlaylistType ]: string } = {
 const RUNNER_JOB_STATES: { [ id in RunnerJobState ]: string } = {
   [RunnerJobState.PROCESSING]: 'Processing',
   [RunnerJobState.COMPLETED]: 'Completed',
+  [RunnerJobState.COMPLETING]: 'Completing',
   [RunnerJobState.PENDING]: 'Pending',
   [RunnerJobState.ERRORED]: 'Errored',
   [RunnerJobState.WAITING_FOR_PARENT_JOB]: 'Waiting for parent job to finish',

server/lib/runners/job-handlers/abstract-job-handler.ts

@@ -1,5 +1,5 @@
 import { throttle } from 'lodash'
-import { retryTransactionWrapper } from '@server/helpers/database-utils'
+import { retryTransactionWrapper, saveInTransactionWithRetries } from '@server/helpers/database-utils'
 import { logger, loggerTagsFactory } from '@server/helpers/logger'
 import { RUNNER_JOBS } from '@server/initializers/constants'
 import { sequelizeTypescript } from '@server/initializers/database'
@@ -12,10 +12,10 @@ import {
   RunnerJobLiveRTMPHLSTranscodingPayload,
   RunnerJobLiveRTMPHLSTranscodingPrivatePayload,
   RunnerJobState,
+  RunnerJobStudioTranscodingPayload,
   RunnerJobSuccessPayload,
   RunnerJobType,
   RunnerJobUpdatePayload,
-  RunnerJobStudioTranscodingPayload,
   RunnerJobVideoStudioTranscodingPrivatePayload,
   RunnerJobVODAudioMergeTranscodingPayload,
   RunnerJobVODAudioMergeTranscodingPrivatePayload,
@@ -139,6 +139,9 @@ export abstract class AbstractJobHandler <C, U extends RunnerJobUpdatePayload, S
   }) {
     const { runnerJob } = options
 
+    runnerJob.state = RunnerJobState.COMPLETING
+    await saveInTransactionWithRetries(runnerJob)
+
     try {
       await this.specificComplete(options)
 
@@ -153,11 +156,7 @@ export abstract class AbstractJobHandler <C, U extends RunnerJobUpdatePayload, S
     runnerJob.progress = null
     runnerJob.finishedAt = new Date()
 
-    await retryTransactionWrapper(() => {
+    await saveInTransactionWithRetries(runnerJob)
-      return sequelizeTypescript.transaction(async transaction => {
-        await runnerJob.save({ transaction })
-      })
-    })
 
     const [ affectedCount ] = await RunnerJobModel.updateDependantJobsOf(runnerJob)
 

shared/models/runners/runner-job-state.model.ts

@@ -6,5 +6,6 @@ export enum RunnerJobState {
   WAITING_FOR_PARENT_JOB = 5,
   CANCELLED = 6,
   PARENT_ERRORED = 7,
-  PARENT_CANCELLED = 8
+  PARENT_CANCELLED = 8,
+  COMPLETING = 9
 }