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

   1 import * as Bull from 'bull'
   2 import {
   3   ActivitypubFollowPayload,
   4   ActivitypubHttpBroadcastPayload,
   5   ActivitypubHttpFetcherPayload, ActivitypubHttpUnicastPayload, EmailPayload,
   6   JobState,
   7   JobType, RefreshPayload, VideoFileImportPayload, VideoImportPayload, VideoRedundancyPayload, VideoTranscodingPayload
   8 } from '../../../shared/models'
   9 import { logger } from '../../helpers/logger'
  10 import { Redis } from '../redis'
  11 import { JOB_ATTEMPTS, JOB_COMPLETED_LIFETIME, JOB_CONCURRENCY, JOB_TTL, REPEAT_JOBS, WEBSERVER } from '../../initializers/constants'
  12 import { processActivityPubHttpBroadcast } from './handlers/activitypub-http-broadcast'
  13 import { processActivityPubHttpFetcher } from './handlers/activitypub-http-fetcher'
  14 import { processActivityPubHttpUnicast } from './handlers/activitypub-http-unicast'
  15 import { processEmail } from './handlers/email'
  16 import { processVideoTranscoding} from './handlers/video-transcoding'
  17 import { processActivityPubFollow } from './handlers/activitypub-follow'
  18 import { processVideoImport} from './handlers/video-import'
  19 import { processVideosViews } from './handlers/video-views'
  20 import { refreshAPObject} from './handlers/activitypub-refresher'
  21 import { processVideoFileImport} from './handlers/video-file-import'
  22 import { processVideoRedundancy} from '@server/lib/job-queue/handlers/video-redundancy'
  23
  24 type CreateJobArgument =
  25   { type: 'activitypub-http-broadcast', payload: ActivitypubHttpBroadcastPayload } |
  26   { type: 'activitypub-http-unicast', payload: ActivitypubHttpUnicastPayload } |
  27   { type: 'activitypub-http-fetcher', payload: ActivitypubHttpFetcherPayload } |
  28   { type: 'activitypub-follow', payload: ActivitypubFollowPayload } |
  29   { type: 'video-file-import', payload: VideoFileImportPayload } |
  30   { type: 'video-transcoding', payload: VideoTranscodingPayload } |
  31   { type: 'email', payload: EmailPayload } |
  32   { type: 'video-import', payload: VideoImportPayload } |
  33   { type: 'activitypub-refresher', payload: RefreshPayload } |
  34   { type: 'videos-views', payload: {} } |
  35   { type: 'video-redundancy', payload: VideoRedundancyPayload }
  36
  37 const handlers: { [id in JobType]: (job: Bull.Job) => Promise<any> } = {
  38   'activitypub-http-broadcast': processActivityPubHttpBroadcast,
  39   'activitypub-http-unicast': processActivityPubHttpUnicast,
  40   'activitypub-http-fetcher': processActivityPubHttpFetcher,
  41   'activitypub-follow': processActivityPubFollow,
  42   'video-file-import': processVideoFileImport,
  43   'video-transcoding': processVideoTranscoding,
  44   'email': processEmail,
  45   'video-import': processVideoImport,
  46   'videos-views': processVideosViews,
  47   'activitypub-refresher': refreshAPObject,
  48   'video-redundancy': processVideoRedundancy
  49 }
  50
  51 const jobTypes: JobType[] = [
  52   'activitypub-follow',
  53   'activitypub-http-broadcast',
  54   'activitypub-http-fetcher',
  55   'activitypub-http-unicast',
  56   'email',
  57   'video-transcoding',
  58   'video-file-import',
  59   'video-import',
  60   'videos-views',
  61   'activitypub-refresher',
  62   'video-redundancy'
  63 ]
  64
  65 class JobQueue {
  66
  67   private static instance: JobQueue
  68
  69   private queues: { [id in JobType]?: Bull.Queue } = {}
  70   private initialized = false
  71   private jobRedisPrefix: string
  72
  73   private constructor () {
  74   }
  75
  76   init () {
  77     // Already initialized
  78     if (this.initialized === true) return
  79     this.initialized = true
  80
  81     this.jobRedisPrefix = 'bull-' + WEBSERVER.HOST
  82     const queueOptions = {
  83       prefix: this.jobRedisPrefix,
  84       redis: Redis.getRedisClientOptions(),
  85       settings: {
  86         maxStalledCount: 10 // transcoding could be long, so jobs can often be interrupted by restarts
  87       }
  88     }
  89
  90     for (const handlerName of Object.keys(handlers)) {
  91       const queue = new Bull(handlerName, queueOptions)
  92       const handler = handlers[handlerName]
  93
  94       queue.process(JOB_CONCURRENCY[handlerName], handler)
  95            .catch(err => logger.error('Error in job queue processor %s.', handlerName, { err }))
  96
  97       queue.on('failed', (job, err) => {
  98         logger.error('Cannot execute job %d in queue %s.', job.id, handlerName, { payload: job.data, err })
  99       })
 100
 101       queue.on('error', err => {
 102         logger.error('Error in job queue %s.', handlerName, { err })
 103       })
 104
 105       this.queues[handlerName] = queue
 106     }
 107
 108     this.addRepeatableJobs()
 109   }
 110
 111   terminate () {
 112     for (const queueName of Object.keys(this.queues)) {
 113       const queue = this.queues[queueName]
 114       queue.close()
 115     }
 116   }
 117
 118   createJob (obj: CreateJobArgument): void {
 119     this.createJobWithPromise(obj)
 120          .catch(err => logger.error('Cannot create job.', { err, obj }))
 121   }
 122
 123   createJobWithPromise (obj: CreateJobArgument) {
 124     const queue = this.queues[obj.type]
 125     if (queue === undefined) {
 126       logger.error('Unknown queue %s: cannot create job.', obj.type)
 127       return
 128     }
 129
 130     const jobArgs: Bull.JobOptions = {
 131       backoff: { delay: 60 * 1000, type: 'exponential' },
 132       attempts: JOB_ATTEMPTS[obj.type],
 133       timeout: JOB_TTL[obj.type]
 134     }
 135
 136     return queue.add(obj.payload, jobArgs)
 137   }
 138
 139   async listForApi (options: {
 140     state: JobState
 141     start: number
 142     count: number
 143     asc?: boolean
 144     jobType: JobType
 145   }): Promise<Bull.Job[]> {
 146     const { state, start, count, asc, jobType } = options
 147     let results: Bull.Job[] = []
 148
 149     const filteredJobTypes = this.filterJobTypes(jobType)
 150
 151     for (const jobType of filteredJobTypes) {
 152       const queue = this.queues[jobType]
 153       if (queue === undefined) {
 154         logger.error('Unknown queue %s to list jobs.', jobType)
 155         continue
 156       }
 157
 158       const jobs = await queue.getJobs([ state ], 0, start + count, asc)
 159       results = results.concat(jobs)
 160     }
 161
 162     results.sort((j1: any, j2: any) => {
 163       if (j1.timestamp < j2.timestamp) return -1
 164       else if (j1.timestamp === j2.timestamp) return 0
 165
 166       return 1
 167     })
 168
 169     if (asc === false) results.reverse()
 170
 171     return results.slice(start, start + count)
 172   }
 173
 174   async count (state: JobState, jobType?: JobType): Promise<number> {
 175     let total = 0
 176
 177     const filteredJobTypes = this.filterJobTypes(jobType)
 178
 179     for (const type of filteredJobTypes) {
 180       const queue = this.queues[type]
 181       if (queue === undefined) {
 182         logger.error('Unknown queue %s to count jobs.', type)
 183         continue
 184       }
 185
 186       const counts = await queue.getJobCounts()
 187
 188       total += counts[state]
 189     }
 190
 191     return total
 192   }
 193
 194   async removeOldJobs () {
 195     for (const key of Object.keys(this.queues)) {
 196       const queue = this.queues[key]
 197       await queue.clean(JOB_COMPLETED_LIFETIME, 'completed')
 198     }
 199   }
 200
 201   private addRepeatableJobs () {
 202     this.queues['videos-views'].add({}, {
 203       repeat: REPEAT_JOBS['videos-views']
 204     }).catch(err => logger.error('Cannot add repeatable job.', { err }))
 205   }
 206
 207   private filterJobTypes (jobType?: JobType) {
 208     if (!jobType) return jobTypes
 209
 210     return jobTypes.filter(t => t === jobType)
 211   }
 212
 213   static get Instance () {
 214     return this.instance || (this.instance = new this())
 215   }
 216 }
 217
 218 // ---------------------------------------------------------------------------
 219
 220 export {
 221   jobTypes,
 222   JobQueue
 223 }