server/lib/job-queue/job-queue.ts

   1 import {
   2   FlowJob,
   3   FlowProducer,
   4   Job,
   5   JobsOptions,
   6   Queue,
   7   QueueEvents,
   8   QueueEventsOptions,
   9   QueueOptions,
  10   Worker,
  11   WorkerOptions
  12 } from 'bullmq'
  13 import { parseDurationToMs } from '@server/helpers/core-utils'
  14 import { jobStates } from '@server/helpers/custom-validators/jobs'
  15 import { CONFIG } from '@server/initializers/config'
  16 import { processVideoRedundancy } from '@server/lib/job-queue/handlers/video-redundancy'
  17 import { pick, timeoutPromise } from '@shared/core-utils'
  18 import {
  19   ActivitypubFollowPayload,
  20   ActivitypubHttpBroadcastPayload,
  21   ActivitypubHttpFetcherPayload,
  22   ActivitypubHttpUnicastPayload,
  23   ActorKeysPayload,
  24   AfterVideoChannelImportPayload,
  25   DeleteResumableUploadMetaFilePayload,
  26   EmailPayload,
  27   FederateVideoPayload,
  28   JobState,
  29   JobType,
  30   ManageVideoTorrentPayload,
  31   MoveObjectStoragePayload,
  32   NotifyPayload,
  33   RefreshPayload,
  34   TranscodingJobBuilderPayload,
  35   VideoChannelImportPayload,
  36   VideoFileImportPayload,
  37   VideoImportPayload,
  38   VideoLiveEndingPayload,
  39   VideoRedundancyPayload,
  40   VideoStudioEditionPayload,
  41   VideoTranscodingPayload
  42 } from '../../../shared/models'
  43 import { logger } from '../../helpers/logger'
  44 import { JOB_ATTEMPTS, JOB_CONCURRENCY, JOB_REMOVAL_OPTIONS, JOB_TTL, REPEAT_JOBS, WEBSERVER } from '../../initializers/constants'
  45 import { Hooks } from '../plugins/hooks'
  46 import { Redis } from '../redis'
  47 import { processActivityPubCleaner } from './handlers/activitypub-cleaner'
  48 import { processActivityPubFollow } from './handlers/activitypub-follow'
  49 import { processActivityPubHttpSequentialBroadcast, processActivityPubParallelHttpBroadcast } from './handlers/activitypub-http-broadcast'
  50 import { processActivityPubHttpFetcher } from './handlers/activitypub-http-fetcher'
  51 import { processActivityPubHttpUnicast } from './handlers/activitypub-http-unicast'
  52 import { refreshAPObject } from './handlers/activitypub-refresher'
  53 import { processActorKeys } from './handlers/actor-keys'
  54 import { processAfterVideoChannelImport } from './handlers/after-video-channel-import'
  55 import { processEmail } from './handlers/email'
  56 import { processFederateVideo } from './handlers/federate-video'
  57 import { processManageVideoTorrent } from './handlers/manage-video-torrent'
  58 import { onMoveToObjectStorageFailure, processMoveToObjectStorage } from './handlers/move-to-object-storage'
  59 import { processNotify } from './handlers/notify'
  60 import { processTranscodingJobBuilder } from './handlers/transcoding-job-builder'
  61 import { processVideoChannelImport } from './handlers/video-channel-import'
  62 import { processVideoFileImport } from './handlers/video-file-import'
  63 import { processVideoImport } from './handlers/video-import'
  64 import { processVideoLiveEnding } from './handlers/video-live-ending'
  65 import { processVideoStudioEdition } from './handlers/video-studio-edition'
  66 import { processVideoTranscoding } from './handlers/video-transcoding'
  67 import { processVideosViewsStats } from './handlers/video-views-stats'
  68
  69 export type CreateJobArgument =
  70   { type: 'activitypub-http-broadcast', payload: ActivitypubHttpBroadcastPayload } |
  71   { type: 'activitypub-http-broadcast-parallel', payload: ActivitypubHttpBroadcastPayload } |
  72   { type: 'activitypub-http-unicast', payload: ActivitypubHttpUnicastPayload } |
  73   { type: 'activitypub-http-fetcher', payload: ActivitypubHttpFetcherPayload } |
  74   { type: 'activitypub-cleaner', payload: {} } |
  75   { type: 'activitypub-follow', payload: ActivitypubFollowPayload } |
  76   { type: 'video-file-import', payload: VideoFileImportPayload } |
  77   { type: 'video-transcoding', payload: VideoTranscodingPayload } |
  78   { type: 'email', payload: EmailPayload } |
  79   { type: 'transcoding-job-builder', payload: TranscodingJobBuilderPayload } |
  80   { type: 'video-import', payload: VideoImportPayload } |
  81   { type: 'activitypub-refresher', payload: RefreshPayload } |
  82   { type: 'videos-views-stats', payload: {} } |
  83   { type: 'video-live-ending', payload: VideoLiveEndingPayload } |
  84   { type: 'actor-keys', payload: ActorKeysPayload } |
  85   { type: 'video-redundancy', payload: VideoRedundancyPayload } |
  86   { type: 'delete-resumable-upload-meta-file', payload: DeleteResumableUploadMetaFilePayload } |
  87   { type: 'video-studio-edition', payload: VideoStudioEditionPayload } |
  88   { type: 'manage-video-torrent', payload: ManageVideoTorrentPayload } |
  89   { type: 'move-to-object-storage', payload: MoveObjectStoragePayload } |
  90   { type: 'video-channel-import', payload: VideoChannelImportPayload } |
  91   { type: 'after-video-channel-import', payload: AfterVideoChannelImportPayload } |
  92   { type: 'notify', payload: NotifyPayload } |
  93   { type: 'move-to-object-storage', payload: MoveObjectStoragePayload } |
  94   { type: 'federate-video', payload: FederateVideoPayload }
  95
  96 export type CreateJobOptions = {
  97   delay?: number
  98   priority?: number
  99 }
 100
 101 const handlers: { [id in JobType]: (job: Job) => Promise<any> } = {
 102   'activitypub-cleaner': processActivityPubCleaner,
 103   'activitypub-follow': processActivityPubFollow,
 104   'activitypub-http-broadcast-parallel': processActivityPubParallelHttpBroadcast,
 105   'activitypub-http-broadcast': processActivityPubHttpSequentialBroadcast,
 106   'activitypub-http-fetcher': processActivityPubHttpFetcher,
 107   'activitypub-http-unicast': processActivityPubHttpUnicast,
 108   'activitypub-refresher': refreshAPObject,
 109   'actor-keys': processActorKeys,
 110   'after-video-channel-import': processAfterVideoChannelImport,
 111   'email': processEmail,
 112   'federate-video': processFederateVideo,
 113   'transcoding-job-builder': processTranscodingJobBuilder,
 114   'manage-video-torrent': processManageVideoTorrent,
 115   'move-to-object-storage': processMoveToObjectStorage,
 116   'notify': processNotify,
 117   'video-channel-import': processVideoChannelImport,
 118   'video-file-import': processVideoFileImport,
 119   'video-import': processVideoImport,
 120   'video-live-ending': processVideoLiveEnding,
 121   'video-redundancy': processVideoRedundancy,
 122   'video-studio-edition': processVideoStudioEdition,
 123   'video-transcoding': processVideoTranscoding,
 124   'videos-views-stats': processVideosViewsStats
 125 }
 126
 127 const errorHandlers: { [id in JobType]?: (job: Job, err: any) => Promise<any> } = {
 128   'move-to-object-storage': onMoveToObjectStorageFailure
 129 }
 130
 131 const jobTypes: JobType[] = [
 132   'activitypub-cleaner',
 133   'activitypub-follow',
 134   'activitypub-http-broadcast-parallel',
 135   'activitypub-http-broadcast',
 136   'activitypub-http-fetcher',
 137   'activitypub-http-unicast',
 138   'activitypub-refresher',
 139   'actor-keys',
 140   'after-video-channel-import',
 141   'email',
 142   'federate-video',
 143   'transcoding-job-builder',
 144   'manage-video-torrent',
 145   'move-to-object-storage',
 146   'notify',
 147   'video-channel-import',
 148   'video-file-import',
 149   'video-import',
 150   'video-live-ending',
 151   'video-redundancy',
 152   'video-studio-edition',
 153   'video-transcoding',
 154   'videos-views-stats'
 155 ]
 156
 157 const silentFailure = new Set<JobType>([ 'activitypub-http-unicast' ])
 158
 159 class JobQueue {
 160
 161   private static instance: JobQueue
 162
 163   private workers: { [id in JobType]?: Worker } = {}
 164   private queues: { [id in JobType]?: Queue } = {}
 165   private queueEvents: { [id in JobType]?: QueueEvents } = {}
 166
 167   private flowProducer: FlowProducer
 168
 169   private initialized = false
 170   private jobRedisPrefix: string
 171
 172   private constructor () {
 173   }
 174
 175   init () {
 176     // Already initialized
 177     if (this.initialized === true) return
 178     this.initialized = true
 179
 180     this.jobRedisPrefix = 'bull-' + WEBSERVER.HOST
 181
 182     for (const handlerName of Object.keys(handlers)) {
 183       this.buildWorker(handlerName)
 184       this.buildQueue(handlerName)
 185       this.buildQueueEvent(handlerName)
 186     }
 187
 188     this.flowProducer = new FlowProducer({
 189       connection: Redis.getRedisClientOptions('FlowProducer'),
 190       prefix: this.jobRedisPrefix
 191     })
 192     this.flowProducer.on('error', err => { logger.error('Error in flow producer', { err }) })
 193
 194     this.addRepeatableJobs()
 195   }
 196
 197   private buildWorker (handlerName: JobType) {
 198     const workerOptions: WorkerOptions = {
 199       autorun: false,
 200       concurrency: this.getJobConcurrency(handlerName),
 201       prefix: this.jobRedisPrefix,
 202       connection: Redis.getRedisClientOptions('Worker'),
 203       maxStalledCount: 10
 204     }
 205
 206     const handler = function (job: Job) {
 207       const timeout = JOB_TTL[handlerName]
 208       const p = handlers[handlerName](job)
 209
 210       if (!timeout) return p
 211
 212       return timeoutPromise(p, timeout)
 213     }
 214
 215     const processor = async (jobArg: Job<any>) => {
 216       const job = await Hooks.wrapObject(jobArg, 'filter:job-queue.process.params', { type: handlerName })
 217
 218       return Hooks.wrapPromiseFun(handler, job, 'filter:job-queue.process.result')
 219     }
 220
 221     const worker = new Worker(handlerName, processor, workerOptions)
 222
 223     worker.on('failed', (job, err) => {
 224       const logLevel = silentFailure.has(handlerName)
 225         ? 'debug'
 226         : 'error'
 227
 228       logger.log(logLevel, 'Cannot execute job %s in queue %s.', job.id, handlerName, { payload: job.data, err })
 229
 230       if (errorHandlers[job.name]) {
 231         errorHandlers[job.name](job, err)
 232           .catch(err => logger.error('Cannot run error handler for job failure %d in queue %s.', job.id, handlerName, { err }))
 233       }
 234     })
 235
 236     worker.on('error', err => { logger.error('Error in job worker %s.', handlerName, { err }) })
 237
 238     this.workers[handlerName] = worker
 239   }
 240
 241   private buildQueue (handlerName: JobType) {
 242     const queueOptions: QueueOptions = {
 243       connection: Redis.getRedisClientOptions('Queue'),
 244       prefix: this.jobRedisPrefix
 245     }
 246
 247     const queue = new Queue(handlerName, queueOptions)
 248     queue.on('error', err => { logger.error('Error in job queue %s.', handlerName, { err }) })
 249
 250     this.queues[handlerName] = queue
 251   }
 252
 253   private buildQueueEvent (handlerName: JobType) {
 254     const queueEventsOptions: QueueEventsOptions = {
 255       autorun: false,
 256       connection: Redis.getRedisClientOptions('QueueEvent'),
 257       prefix: this.jobRedisPrefix
 258     }
 259
 260     const queueEvents = new QueueEvents(handlerName, queueEventsOptions)
 261     queueEvents.on('error', err => { logger.error('Error in job queue events %s.', handlerName, { err }) })
 262
 263     this.queueEvents[handlerName] = queueEvents
 264   }
 265
 266   // ---------------------------------------------------------------------------
 267
 268   async terminate () {
 269     const promises = Object.keys(this.workers)
 270       .map(handlerName => {
 271         const worker: Worker = this.workers[handlerName]
 272         const queue: Queue = this.queues[handlerName]
 273         const queueEvent: QueueEvents = this.queueEvents[handlerName]
 274
 275         return Promise.all([
 276           worker.close(false),
 277           queue.close(),
 278           queueEvent.close()
 279         ])
 280       })
 281
 282     return Promise.all(promises)
 283   }
 284
 285   start () {
 286     const promises = Object.keys(this.workers)
 287       .map(handlerName => {
 288         const worker: Worker = this.workers[handlerName]
 289         const queueEvent: QueueEvents = this.queueEvents[handlerName]
 290
 291         return Promise.all([
 292           worker.run(),
 293           queueEvent.run()
 294         ])
 295       })
 296
 297     return Promise.all(promises)
 298   }
 299
 300   async pause () {
 301     for (const handlerName of Object.keys(this.workers)) {
 302       const worker: Worker = this.workers[handlerName]
 303
 304       await worker.pause()
 305     }
 306   }
 307
 308   resume () {
 309     for (const handlerName of Object.keys(this.workers)) {
 310       const worker: Worker = this.workers[handlerName]
 311
 312       worker.resume()
 313     }
 314   }
 315
 316   // ---------------------------------------------------------------------------
 317
 318   createJobAsync (options: CreateJobArgument & CreateJobOptions): void {
 319     this.createJob(options)
 320         .catch(err => logger.error('Cannot create job.', { err, options }))
 321   }
 322
 323   createJob (options: CreateJobArgument & CreateJobOptions) {
 324     const queue: Queue = this.queues[options.type]
 325     if (queue === undefined) {
 326       logger.error('Unknown queue %s: cannot create job.', options.type)
 327       return
 328     }
 329
 330     const jobOptions = this.buildJobOptions(options.type as JobType, pick(options, [ 'priority', 'delay' ]))
 331
 332     return queue.add('job', options.payload, jobOptions)
 333   }
 334
 335   createSequentialJobFlow (...jobs: ((CreateJobArgument & CreateJobOptions) | undefined)[]) {
 336     let lastJob: FlowJob
 337
 338     for (const job of jobs) {
 339       if (!job) continue
 340
 341       lastJob = {
 342         ...this.buildJobFlowOption(job),
 343
 344         children: lastJob
 345           ? [ lastJob ]
 346           : []
 347       }
 348     }
 349
 350     return this.flowProducer.add(lastJob)
 351   }
 352
 353   createJobWithChildren (parent: CreateJobArgument & CreateJobOptions, children: (CreateJobArgument & CreateJobOptions)[]) {
 354     return this.flowProducer.add({
 355       ...this.buildJobFlowOption(parent),
 356
 357       children: children.map(c => this.buildJobFlowOption(c))
 358     })
 359   }
 360
 361   private buildJobFlowOption (job: CreateJobArgument & CreateJobOptions): FlowJob {
 362     return {
 363       name: 'job',
 364       data: job.payload,
 365       queueName: job.type,
 366       opts: this.buildJobOptions(job.type as JobType, pick(job, [ 'priority', 'delay' ]))
 367     }
 368   }
 369
 370   private buildJobOptions (type: JobType, options: CreateJobOptions = {}): JobsOptions {
 371     return {
 372       backoff: { delay: 60 * 1000, type: 'exponential' },
 373       attempts: JOB_ATTEMPTS[type],
 374       priority: options.priority,
 375       delay: options.delay,
 376
 377       ...this.buildJobRemovalOptions(type)
 378     }
 379   }
 380
 381   // ---------------------------------------------------------------------------
 382
 383   async listForApi (options: {
 384     state?: JobState
 385     start: number
 386     count: number
 387     asc?: boolean
 388     jobType: JobType
 389   }): Promise<Job[]> {
 390     const { state, start, count, asc, jobType } = options
 391
 392     const states = this.buildStateFilter(state)
 393     const filteredJobTypes = this.buildTypeFilter(jobType)
 394
 395     let results: Job[] = []
 396
 397     for (const jobType of filteredJobTypes) {
 398       const queue: Queue = this.queues[jobType]
 399
 400       if (queue === undefined) {
 401         logger.error('Unknown queue %s to list jobs.', jobType)
 402         continue
 403       }
 404
 405       const jobs = await queue.getJobs(states, 0, start + count, asc)
 406       results = results.concat(jobs)
 407     }
 408
 409     results.sort((j1: any, j2: any) => {
 410       if (j1.timestamp < j2.timestamp) return -1
 411       else if (j1.timestamp === j2.timestamp) return 0
 412
 413       return 1
 414     })
 415
 416     if (asc === false) results.reverse()
 417
 418     return results.slice(start, start + count)
 419   }
 420
 421   async count (state: JobState, jobType?: JobType): Promise<number> {
 422     const states = state ? [ state ] : jobStates
 423     const filteredJobTypes = this.buildTypeFilter(jobType)
 424
 425     let total = 0
 426
 427     for (const type of filteredJobTypes) {
 428       const queue = this.queues[type]
 429       if (queue === undefined) {
 430         logger.error('Unknown queue %s to count jobs.', type)
 431         continue
 432       }
 433
 434       const counts = await queue.getJobCounts()
 435
 436       for (const s of states) {
 437         total += counts[s]
 438       }
 439     }
 440
 441     return total
 442   }
 443
 444   private buildStateFilter (state?: JobState) {
 445     if (!state) return jobStates
 446
 447     const states = [ state ]
 448
 449     // Include parent if filtering on waiting
 450     if (state === 'waiting') states.push('waiting-children')
 451
 452     return states
 453   }
 454
 455   private buildTypeFilter (jobType?: JobType) {
 456     if (!jobType) return jobTypes
 457
 458     return jobTypes.filter(t => t === jobType)
 459   }
 460
 461   async getStats () {
 462     const promises = jobTypes.map(async t => ({ jobType: t, counts: await this.queues[t].getJobCounts() }))
 463
 464     return Promise.all(promises)
 465   }
 466
 467   // ---------------------------------------------------------------------------
 468
 469   async removeOldJobs () {
 470     for (const key of Object.keys(this.queues)) {
 471       const queue: Queue = this.queues[key]
 472       await queue.clean(parseDurationToMs('7 days'), 1000, 'completed')
 473       await queue.clean(parseDurationToMs('7 days'), 1000, 'failed')
 474     }
 475   }
 476
 477   private addRepeatableJobs () {
 478     this.queues['videos-views-stats'].add('job', {}, {
 479       repeat: REPEAT_JOBS['videos-views-stats'],
 480
 481       ...this.buildJobRemovalOptions('videos-views-stats')
 482     }).catch(err => logger.error('Cannot add repeatable job.', { err }))
 483
 484     if (CONFIG.FEDERATION.VIDEOS.CLEANUP_REMOTE_INTERACTIONS) {
 485       this.queues['activitypub-cleaner'].add('job', {}, {
 486         repeat: REPEAT_JOBS['activitypub-cleaner'],
 487
 488         ...this.buildJobRemovalOptions('activitypub-cleaner')
 489       }).catch(err => logger.error('Cannot add repeatable job.', { err }))
 490     }
 491   }
 492
 493   private getJobConcurrency (jobType: JobType) {
 494     if (jobType === 'video-transcoding') return CONFIG.TRANSCODING.CONCURRENCY
 495     if (jobType === 'video-import') return CONFIG.IMPORT.VIDEOS.CONCURRENCY
 496
 497     return JOB_CONCURRENCY[jobType]
 498   }
 499
 500   private buildJobRemovalOptions (queueName: string) {
 501     return {
 502       removeOnComplete: {
 503         // Wants seconds
 504         age: (JOB_REMOVAL_OPTIONS.SUCCESS[queueName] || JOB_REMOVAL_OPTIONS.SUCCESS.DEFAULT) / 1000,
 505
 506         count: JOB_REMOVAL_OPTIONS.COUNT
 507       },
 508       removeOnFail: {
 509         // Wants seconds
 510         age: (JOB_REMOVAL_OPTIONS.FAILURE[queueName] || JOB_REMOVAL_OPTIONS.FAILURE.DEFAULT) / 1000,
 511
 512         count: JOB_REMOVAL_OPTIONS.COUNT / 1000
 513       }
 514     }
 515   }
 516
 517   static get Instance () {
 518     return this.instance || (this.instance = new this())
 519   }
 520 }
 521
 522 // ---------------------------------------------------------------------------
 523
 524 export {
 525   jobTypes,
 526   JobQueue
 527 }