1 import { throttle } from 'lodash'
2 import { retryTransactionWrapper, 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 retryTransactionWrapper(() => {
128 return sequelizeTypescript.transaction(async transaction => {
129 return runnerJob.save({ transaction })
134 // ---------------------------------------------------------------------------
136 async complete (options: {
137 runnerJob: MRunnerJob
140 const { runnerJob } = options
142 runnerJob.state = RunnerJobState.COMPLETING
143 await saveInTransactionWithRetries(runnerJob)
146 await this.specificComplete(options)
148 runnerJob.state = RunnerJobState.COMPLETED
150 logger.error('Cannot complete runner job', { err, ...this.lTags(runnerJob.id, runnerJob.type) })
152 runnerJob.state = RunnerJobState.ERRORED
153 runnerJob.error = err.message
156 runnerJob.progress = null
157 runnerJob.finishedAt = new Date()
159 await saveInTransactionWithRetries(runnerJob)
161 const [ affectedCount ] = await RunnerJobModel.updateDependantJobsOf(runnerJob)
163 if (affectedCount !== 0) PeerTubeSocket.Instance.sendAvailableJobsPingToRunners()
166 protected abstract specificComplete (options: {
167 runnerJob: MRunnerJob
169 }): Promise<void> | void
171 // ---------------------------------------------------------------------------
173 async cancel (options: {
174 runnerJob: MRunnerJob
177 const { runnerJob, fromParent } = options
179 await this.specificCancel(options)
181 const cancelState = fromParent
182 ? RunnerJobState.PARENT_CANCELLED
183 : RunnerJobState.CANCELLED
185 runnerJob.setToErrorOrCancel(cancelState)
187 await retryTransactionWrapper(() => {
188 return sequelizeTypescript.transaction(async transaction => {
189 await runnerJob.save({ transaction })
193 const children = await RunnerJobModel.listChildrenOf(runnerJob)
194 for (const child of children) {
195 logger.info(`Cancelling child job ${child.uuid} of ${runnerJob.uuid} because of parent cancel`, this.lTags(child.uuid))
197 await this.cancel({ runnerJob: child, fromParent: true })
201 protected abstract specificCancel (options: {
202 runnerJob: MRunnerJob
203 }): Promise<void> | void
205 // ---------------------------------------------------------------------------
207 protected abstract isAbortSupported (): boolean
209 async abort (options: {
210 runnerJob: MRunnerJob
212 const { runnerJob } = options
214 if (this.isAbortSupported() !== true) {
215 return this.error({ runnerJob, message: 'Job has been aborted but it is not supported by this job type' })
218 await this.specificAbort(options)
220 runnerJob.resetToPending()
222 await retryTransactionWrapper(() => {
223 return sequelizeTypescript.transaction(async transaction => {
224 await runnerJob.save({ transaction })
229 protected setAbortState (runnerJob: MRunnerJob) {
230 runnerJob.resetToPending()
233 protected abstract specificAbort (options: {
234 runnerJob: MRunnerJob
235 }): Promise<void> | void
237 // ---------------------------------------------------------------------------
239 async error (options: {
240 runnerJob: MRunnerJob
244 const { runnerJob, message, fromParent } = options
246 const errorState = fromParent
247 ? RunnerJobState.PARENT_ERRORED
248 : RunnerJobState.ERRORED
250 const nextState = errorState === RunnerJobState.ERRORED && this.isAbortSupported() && runnerJob.failures < RUNNER_JOBS.MAX_FAILURES
251 ? RunnerJobState.PENDING
254 await this.specificError({ ...options, nextState })
256 if (nextState === errorState) {
257 runnerJob.setToErrorOrCancel(nextState)
258 runnerJob.error = message
260 runnerJob.resetToPending()
263 await retryTransactionWrapper(() => {
264 return sequelizeTypescript.transaction(async transaction => {
265 await runnerJob.save({ transaction })
269 if (runnerJob.state === errorState) {
270 const children = await RunnerJobModel.listChildrenOf(runnerJob)
272 for (const child of children) {
273 logger.info(`Erroring child job ${child.uuid} of ${runnerJob.uuid} because of parent error`, this.lTags(child.uuid))
275 await this.error({ runnerJob: child, message: 'Parent error', fromParent: true })
280 protected abstract specificError (options: {
281 runnerJob: MRunnerJob
283 nextState: RunnerJobState
284 }): Promise<void> | void