1 import { throttle } from 'lodash'
2 import { saveInTransactionWithRetries } from '@server/helpers/database-utils'
3 import { logger, loggerTagsFactory } from '@server/helpers/logger'
4 import { RUNNER_JOBS } from '@server/initializers/constants'
5 import { sequelizeTypescript } from '@server/initializers/database'
6 import { PeerTubeSocket } from '@server/lib/peertube-socket'
7 import { RunnerJobModel } from '@server/models/runner/runner-job'
8 import { setAsUpdated } from '@server/models/shared'
9 import { MRunnerJob } from '@server/types/models/runners'
10 import { pick } from '@shared/core-utils'
12 RunnerJobLiveRTMPHLSTranscodingPayload,
13 RunnerJobLiveRTMPHLSTranscodingPrivatePayload,
15 RunnerJobStudioTranscodingPayload,
16 RunnerJobSuccessPayload,
18 RunnerJobUpdatePayload,
19 RunnerJobVideoStudioTranscodingPrivatePayload,
20 RunnerJobVODAudioMergeTranscodingPayload,
21 RunnerJobVODAudioMergeTranscodingPrivatePayload,
22 RunnerJobVODHLSTranscodingPayload,
23 RunnerJobVODHLSTranscodingPrivatePayload,
24 RunnerJobVODWebVideoTranscodingPayload,
25 RunnerJobVODWebVideoTranscodingPrivatePayload
26 } from '@shared/models'
28 type CreateRunnerJobArg =
30 type: Extract<RunnerJobType, 'vod-web-video-transcoding'>
31 payload: RunnerJobVODWebVideoTranscodingPayload
32 privatePayload: RunnerJobVODWebVideoTranscodingPrivatePayload
35 type: Extract<RunnerJobType, 'vod-hls-transcoding'>
36 payload: RunnerJobVODHLSTranscodingPayload
37 privatePayload: RunnerJobVODHLSTranscodingPrivatePayload
40 type: Extract<RunnerJobType, 'vod-audio-merge-transcoding'>
41 payload: RunnerJobVODAudioMergeTranscodingPayload
42 privatePayload: RunnerJobVODAudioMergeTranscodingPrivatePayload
45 type: Extract<RunnerJobType, 'live-rtmp-hls-transcoding'>
46 payload: RunnerJobLiveRTMPHLSTranscodingPayload
47 privatePayload: RunnerJobLiveRTMPHLSTranscodingPrivatePayload
50 type: Extract<RunnerJobType, 'video-studio-transcoding'>
51 payload: RunnerJobStudioTranscodingPayload
52 privatePayload: RunnerJobVideoStudioTranscodingPrivatePayload
55 export abstract class AbstractJobHandler <C, U extends RunnerJobUpdatePayload, S extends RunnerJobSuccessPayload> {
57 protected readonly lTags = loggerTagsFactory('runner')
59 static setJobAsUpdatedThrottled = throttle(setAsUpdated, 2000)
61 // ---------------------------------------------------------------------------
63 abstract create (options: C): Promise<MRunnerJob>
65 protected async createRunnerJob (options: CreateRunnerJobArg & {
68 dependsOnRunnerJob?: MRunnerJob
69 }): Promise<MRunnerJob> {
70 const { priority, dependsOnRunnerJob } = options
72 logger.debug('Creating runner job', { options, ...this.lTags(options.type) })
74 const runnerJob = new RunnerJobModel({
75 ...pick(options, [ 'type', 'payload', 'privatePayload' ]),
77 uuid: options.jobUUID,
79 state: dependsOnRunnerJob
80 ? RunnerJobState.WAITING_FOR_PARENT_JOB
81 : RunnerJobState.PENDING,
83 dependsOnRunnerJobId: dependsOnRunnerJob?.id,
88 const job = await sequelizeTypescript.transaction(async transaction => {
89 return runnerJob.save({ transaction })
92 if (runnerJob.state === RunnerJobState.PENDING) {
93 PeerTubeSocket.Instance.sendAvailableJobsPingToRunners()
99 // ---------------------------------------------------------------------------
101 protected abstract specificUpdate (options: {
102 runnerJob: MRunnerJob
104 }): Promise<void> | void
106 async update (options: {
107 runnerJob: MRunnerJob
111 const { runnerJob, progress } = options
113 await this.specificUpdate(options)
115 if (progress) runnerJob.progress = progress
117 if (!runnerJob.changed()) {
119 await AbstractJobHandler.setJobAsUpdatedThrottled({ sequelize: sequelizeTypescript, table: 'runnerJob', id: runnerJob.id })
121 logger.warn('Cannot set remote job as updated', { err, ...this.lTags(runnerJob.id, runnerJob.type) })
127 await saveInTransactionWithRetries(runnerJob)
130 // ---------------------------------------------------------------------------
132 async complete (options: {
133 runnerJob: MRunnerJob
136 const { runnerJob } = options
138 runnerJob.state = RunnerJobState.COMPLETING
139 await saveInTransactionWithRetries(runnerJob)
142 await this.specificComplete(options)
144 runnerJob.state = RunnerJobState.COMPLETED
146 logger.error('Cannot complete runner job', { err, ...this.lTags(runnerJob.id, runnerJob.type) })
148 runnerJob.state = RunnerJobState.ERRORED
149 runnerJob.error = err.message
152 runnerJob.progress = null
153 runnerJob.finishedAt = new Date()
155 await saveInTransactionWithRetries(runnerJob)
157 const [ affectedCount ] = await RunnerJobModel.updateDependantJobsOf(runnerJob)
159 if (affectedCount !== 0) PeerTubeSocket.Instance.sendAvailableJobsPingToRunners()
162 protected abstract specificComplete (options: {
163 runnerJob: MRunnerJob
165 }): Promise<void> | void
167 // ---------------------------------------------------------------------------
169 async cancel (options: {
170 runnerJob: MRunnerJob
173 const { runnerJob, fromParent } = options
175 await this.specificCancel(options)
177 const cancelState = fromParent
178 ? RunnerJobState.PARENT_CANCELLED
179 : RunnerJobState.CANCELLED
181 runnerJob.setToErrorOrCancel(cancelState)
183 await saveInTransactionWithRetries(runnerJob)
185 const children = await RunnerJobModel.listChildrenOf(runnerJob)
186 for (const child of children) {
187 logger.info(`Cancelling child job ${child.uuid} of ${runnerJob.uuid} because of parent cancel`, this.lTags(child.uuid))
189 await this.cancel({ runnerJob: child, fromParent: true })
193 protected abstract specificCancel (options: {
194 runnerJob: MRunnerJob
195 }): Promise<void> | void
197 // ---------------------------------------------------------------------------
199 protected abstract isAbortSupported (): boolean
201 async abort (options: {
202 runnerJob: MRunnerJob
204 const { runnerJob } = options
206 if (this.isAbortSupported() !== true) {
207 return this.error({ runnerJob, message: 'Job has been aborted but it is not supported by this job type' })
210 await this.specificAbort(options)
212 runnerJob.resetToPending()
214 await saveInTransactionWithRetries(runnerJob)
217 protected setAbortState (runnerJob: MRunnerJob) {
218 runnerJob.resetToPending()
221 protected abstract specificAbort (options: {
222 runnerJob: MRunnerJob
223 }): Promise<void> | void
225 // ---------------------------------------------------------------------------
227 async error (options: {
228 runnerJob: MRunnerJob
232 const { runnerJob, message, fromParent } = options
234 const errorState = fromParent
235 ? RunnerJobState.PARENT_ERRORED
236 : RunnerJobState.ERRORED
238 const nextState = errorState === RunnerJobState.ERRORED && this.isAbortSupported() && runnerJob.failures < RUNNER_JOBS.MAX_FAILURES
239 ? RunnerJobState.PENDING
242 await this.specificError({ ...options, nextState })
244 if (nextState === errorState) {
245 runnerJob.setToErrorOrCancel(nextState)
246 runnerJob.error = message
248 runnerJob.resetToPending()
251 await saveInTransactionWithRetries(runnerJob)
253 if (runnerJob.state === errorState) {
254 const children = await RunnerJobModel.listChildrenOf(runnerJob)
256 for (const child of children) {
257 logger.info(`Erroring child job ${child.uuid} of ${runnerJob.uuid} because of parent error`, this.lTags(child.uuid))
259 await this.error({ runnerJob: child, message: 'Parent error', fromParent: true })
264 protected abstract specificError (options: {
265 runnerJob: MRunnerJob
267 nextState: RunnerJobState
268 }): Promise<void> | void