X-Git-Url: https://git.immae.eu/?a=blobdiff_plain;f=server%2Flib%2Fjob-queue%2Fjob-queue.ts;h=42e8347b1f663650c5aed7042c5e02cbd1b9ffcc;hb=ac27887774e63d99f4e227fbe18846f143cc4b3c;hp=d8d64caaf348bae8e464cfb58bbd1d53f26e46dd;hpb=8dc8a34ee8428e7657414115d1c137592efa174d;p=github%2FChocobozzz%2FPeerTube.git diff --git a/server/lib/job-queue/job-queue.ts b/server/lib/job-queue/job-queue.ts index d8d64caaf..42e8347b1 100644 --- a/server/lib/job-queue/job-queue.ts +++ b/server/lib/job-queue/job-queue.ts @@ -1,30 +1,45 @@ import * as Bull from 'bull' +import { jobStates } from '@server/helpers/custom-validators/jobs' +import { CONFIG } from '@server/initializers/config' +import { processVideoRedundancy } from '@server/lib/job-queue/handlers/video-redundancy' import { ActivitypubFollowPayload, ActivitypubHttpBroadcastPayload, - ActivitypubHttpFetcherPayload, ActivitypubHttpUnicastPayload, EmailPayload, + ActivitypubHttpFetcherPayload, + ActivitypubHttpUnicastPayload, + ActorKeysPayload, + EmailPayload, JobState, - JobType, RefreshPayload, VideoFileImportPayload, VideoImportPayload, VideoRedundancyPayload, VideoTranscodingPayload + JobType, + RefreshPayload, + VideoFileImportPayload, + VideoImportPayload, + VideoLiveEndingPayload, + VideoRedundancyPayload, + VideoTranscodingPayload } from '../../../shared/models' import { logger } from '../../helpers/logger' -import { Redis } from '../redis' import { JOB_ATTEMPTS, JOB_COMPLETED_LIFETIME, JOB_CONCURRENCY, JOB_TTL, REPEAT_JOBS, WEBSERVER } from '../../initializers/constants' +import { Redis } from '../redis' +import { processActivityPubCleaner } from './handlers/activitypub-cleaner' +import { processActivityPubFollow } from './handlers/activitypub-follow' import { processActivityPubHttpBroadcast } from './handlers/activitypub-http-broadcast' import { processActivityPubHttpFetcher } from './handlers/activitypub-http-fetcher' import { processActivityPubHttpUnicast } from './handlers/activitypub-http-unicast' +import { refreshAPObject } from './handlers/activitypub-refresher' +import { processActorKeys } from './handlers/actor-keys' import { processEmail } from './handlers/email' -import { processVideoTranscoding} from './handlers/video-transcoding' -import { processActivityPubFollow } from './handlers/activitypub-follow' -import { processVideoImport} from './handlers/video-import' +import { processVideoFileImport } from './handlers/video-file-import' +import { processVideoImport } from './handlers/video-import' +import { processVideoLiveEnding } from './handlers/video-live-ending' +import { processVideoTranscoding } from './handlers/video-transcoding' import { processVideosViews } from './handlers/video-views' -import { refreshAPObject} from './handlers/activitypub-refresher' -import { processVideoFileImport} from './handlers/video-file-import' -import { processVideoRedundancy} from '@server/lib/job-queue/handlers/video-redundancy' type CreateJobArgument = { type: 'activitypub-http-broadcast', payload: ActivitypubHttpBroadcastPayload } | { type: 'activitypub-http-unicast', payload: ActivitypubHttpUnicastPayload } | { type: 'activitypub-http-fetcher', payload: ActivitypubHttpFetcherPayload } | + { type: 'activitypub-http-cleaner', payload: {} } | { type: 'activitypub-follow', payload: ActivitypubFollowPayload } | { type: 'video-file-import', payload: VideoFileImportPayload } | { type: 'video-transcoding', payload: VideoTranscodingPayload } | @@ -32,12 +47,20 @@ type CreateJobArgument = { type: 'video-import', payload: VideoImportPayload } | { type: 'activitypub-refresher', payload: RefreshPayload } | { type: 'videos-views', payload: {} } | + { type: 'video-live-ending', payload: VideoLiveEndingPayload } | + { type: 'actor-keys', payload: ActorKeysPayload } | { type: 'video-redundancy', payload: VideoRedundancyPayload } +type CreateJobOptions = { + delay?: number + priority?: number +} + const handlers: { [id in JobType]: (job: Bull.Job) => Promise } = { 'activitypub-http-broadcast': processActivityPubHttpBroadcast, 'activitypub-http-unicast': processActivityPubHttpUnicast, 'activitypub-http-fetcher': processActivityPubHttpFetcher, + 'activitypub-cleaner': processActivityPubCleaner, 'activitypub-follow': processActivityPubFollow, 'video-file-import': processVideoFileImport, 'video-transcoding': processVideoTranscoding, @@ -45,6 +68,8 @@ const handlers: { [id in JobType]: (job: Bull.Job) => Promise } = { 'video-import': processVideoImport, 'videos-views': processVideosViews, 'activitypub-refresher': refreshAPObject, + 'video-live-ending': processVideoLiveEnding, + 'actor-keys': processActorKeys, 'video-redundancy': processVideoRedundancy } @@ -53,13 +78,16 @@ const jobTypes: JobType[] = [ 'activitypub-http-broadcast', 'activitypub-http-fetcher', 'activitypub-http-unicast', + 'activitypub-cleaner', 'email', 'video-transcoding', 'video-file-import', 'video-import', 'videos-views', 'activitypub-refresher', - 'video-redundancy' + 'video-redundancy', + 'actor-keys', + 'video-live-ending' ] class JobQueue { @@ -87,11 +115,11 @@ class JobQueue { } } - for (const handlerName of Object.keys(handlers)) { + for (const handlerName of (Object.keys(handlers) as JobType[])) { const queue = new Bull(handlerName, queueOptions) const handler = handlers[handlerName] - queue.process(JOB_CONCURRENCY[handlerName], handler) + queue.process(this.getJobConcurrency(handlerName), handler) .catch(err => logger.error('Error in job queue processor %s.', handlerName, { err })) queue.on('failed', (job, err) => { @@ -115,12 +143,12 @@ class JobQueue { } } - createJob (obj: CreateJobArgument): void { - this.createJobWithPromise(obj) - .catch(err => logger.error('Cannot create job.', { err, obj })) + createJob (obj: CreateJobArgument, options: CreateJobOptions = {}): void { + this.createJobWithPromise(obj, options) + .catch(err => logger.error('Cannot create job.', { err, obj })) } - createJobWithPromise (obj: CreateJobArgument) { + createJobWithPromise (obj: CreateJobArgument, options: CreateJobOptions = {}) { const queue = this.queues[obj.type] if (queue === undefined) { logger.error('Unknown queue %s: cannot create job.', obj.type) @@ -130,20 +158,24 @@ class JobQueue { const jobArgs: Bull.JobOptions = { backoff: { delay: 60 * 1000, type: 'exponential' }, attempts: JOB_ATTEMPTS[obj.type], - timeout: JOB_TTL[obj.type] + timeout: JOB_TTL[obj.type], + priority: options.priority, + delay: options.delay } return queue.add(obj.payload, jobArgs) } async listForApi (options: { - state: JobState + state?: JobState start: number count: number asc?: boolean jobType: JobType }): Promise { const { state, start, count, asc, jobType } = options + + const states = state ? [ state ] : jobStates let results: Bull.Job[] = [] const filteredJobTypes = this.filterJobTypes(jobType) @@ -155,7 +187,7 @@ class JobQueue { continue } - const jobs = await queue.getJobs([ state ], 0, start + count, asc) + const jobs = await queue.getJobs(states, 0, start + count, asc) results = results.concat(jobs) } @@ -172,6 +204,7 @@ class JobQueue { } async count (state: JobState, jobType?: JobType): Promise { + const states = state ? [ state ] : jobStates let total = 0 const filteredJobTypes = this.filterJobTypes(jobType) @@ -185,7 +218,9 @@ class JobQueue { const counts = await queue.getJobCounts() - total += counts[state] + for (const s of states) { + total += counts[s] + } } return total @@ -202,6 +237,12 @@ class JobQueue { this.queues['videos-views'].add({}, { repeat: REPEAT_JOBS['videos-views'] }).catch(err => logger.error('Cannot add repeatable job.', { err })) + + if (CONFIG.FEDERATION.VIDEOS.CLEANUP_REMOTE_INTERACTIONS) { + this.queues['activitypub-cleaner'].add({}, { + repeat: REPEAT_JOBS['activitypub-cleaner'] + }).catch(err => logger.error('Cannot add repeatable job.', { err })) + } } private filterJobTypes (jobType?: JobType) { @@ -210,6 +251,13 @@ class JobQueue { return jobTypes.filter(t => t === jobType) } + private getJobConcurrency (jobType: JobType) { + if (jobType === 'video-transcoding') return CONFIG.TRANSCODING.CONCURRENCY + if (jobType === 'video-import') return CONFIG.IMPORT.VIDEOS.CONCURRENCY + + return JOB_CONCURRENCY[jobType] + } + static get Instance () { return this.instance || (this.instance = new this()) }