ActivitypubHttpFetcherPayload,
ActivitypubHttpUnicastPayload,
ActorKeysPayload,
ActivitypubHttpFetcherPayload,
ActivitypubHttpUnicastPayload,
ActorKeysPayload,
-import { JOB_ATTEMPTS, JOB_COMPLETED_LIFETIME, JOB_CONCURRENCY, JOB_TTL, REPEAT_JOBS, WEBSERVER } from '../../initializers/constants'
+import {
+ JOB_ATTEMPTS,
+ JOB_CONCURRENCY,
+ JOB_REMOVAL_OPTIONS,
+ JOB_TTL,
+ REPEAT_JOBS,
+ WEBSERVER
+} from '../../initializers/constants'
import { processActivityPubCleaner } from './handlers/activitypub-cleaner'
import { processActivityPubFollow } from './handlers/activitypub-follow'
import { processActivityPubCleaner } from './handlers/activitypub-cleaner'
import { processActivityPubFollow } from './handlers/activitypub-follow'
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 { 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 { processFederateVideo } from './handlers/federate-video'
import { processManageVideoTorrent } from './handlers/manage-video-torrent'
import { onMoveToObjectStorageFailure, processMoveToObjectStorage } from './handlers/move-to-object-storage'
import { processNotify } from './handlers/notify'
import { processEmail } from './handlers/email'
import { processFederateVideo } from './handlers/federate-video'
import { processManageVideoTorrent } from './handlers/manage-video-torrent'
import { onMoveToObjectStorageFailure, processMoveToObjectStorage } from './handlers/move-to-object-storage'
import { processNotify } from './handlers/notify'
import { processVideoFileImport } from './handlers/video-file-import'
import { processVideoImport } from './handlers/video-import'
import { processVideoLiveEnding } from './handlers/video-live-ending'
import { processVideoStudioEdition } from './handlers/video-studio-edition'
import { processVideoTranscoding } from './handlers/video-transcoding'
import { processVideosViewsStats } from './handlers/video-views-stats'
import { processVideoFileImport } from './handlers/video-file-import'
import { processVideoImport } from './handlers/video-import'
import { processVideoLiveEnding } from './handlers/video-live-ending'
import { processVideoStudioEdition } from './handlers/video-studio-edition'
import { processVideoTranscoding } from './handlers/video-transcoding'
import { processVideosViewsStats } from './handlers/video-views-stats'
export type CreateJobArgument =
{ type: 'activitypub-http-broadcast', payload: ActivitypubHttpBroadcastPayload } |
export type CreateJobArgument =
{ type: 'activitypub-http-broadcast', payload: ActivitypubHttpBroadcastPayload } |
{ type: 'delete-resumable-upload-meta-file', payload: DeleteResumableUploadMetaFilePayload } |
{ type: 'video-studio-edition', payload: VideoStudioEditionPayload } |
{ type: 'manage-video-torrent', payload: ManageVideoTorrentPayload } |
{ type: 'delete-resumable-upload-meta-file', payload: DeleteResumableUploadMetaFilePayload } |
{ type: 'video-studio-edition', payload: VideoStudioEditionPayload } |
{ type: 'manage-video-torrent', payload: ManageVideoTorrentPayload } |
+ { type: 'move-to-object-storage', payload: MoveObjectStoragePayload } |
+ { type: 'video-channel-import', payload: VideoChannelImportPayload } |
+ { type: 'after-video-channel-import', payload: AfterVideoChannelImportPayload } |
{ type: 'notify', payload: NotifyPayload } |
{ type: 'move-to-object-storage', payload: MoveObjectStoragePayload } |
{ type: 'federate-video', payload: FederateVideoPayload }
{ type: 'notify', payload: NotifyPayload } |
{ type: 'move-to-object-storage', payload: MoveObjectStoragePayload } |
{ type: 'federate-video', payload: FederateVideoPayload }
- 'activitypub-http-broadcast': processActivityPubHttpBroadcast,
- 'activitypub-http-broadcast-parallel': processActivityPubHttpBroadcast,
+ 'activitypub-http-broadcast': processActivityPubHttpSequentialBroadcast,
+ 'activitypub-http-broadcast-parallel': processActivityPubParallelHttpBroadcast,
'activitypub-http-unicast': processActivityPubHttpUnicast,
'activitypub-http-fetcher': processActivityPubHttpFetcher,
'activitypub-cleaner': processActivityPubCleaner,
'activitypub-http-unicast': processActivityPubHttpUnicast,
'activitypub-http-fetcher': processActivityPubHttpFetcher,
'activitypub-cleaner': processActivityPubCleaner,
'video-redundancy': processVideoRedundancy,
'move-to-object-storage': processMoveToObjectStorage,
'manage-video-torrent': processManageVideoTorrent,
'video-redundancy': processVideoRedundancy,
'move-to-object-storage': processMoveToObjectStorage,
'manage-video-torrent': processManageVideoTorrent,
+ 'video-channel-import': processVideoChannelImport,
+ 'after-video-channel-import': processAfterVideoChannelImport,
+ 'notify': processNotify,
- for (const handlerName of (Object.keys(handlers) as JobType[])) {
- this.buildWorker(handlerName, produceOnly)
+ for (const handlerName of Object.keys(handlers)) {
+ this.buildWorker(handlerName)
- this.buildQueueScheduler(handlerName, produceOnly)
- this.buildQueueEvent(handlerName, produceOnly)
+ this.buildQueueScheduler(handlerName)
+ this.buildQueueEvent(handlerName)
concurrency: this.getJobConcurrency(handlerName),
prefix: this.jobRedisPrefix,
concurrency: this.getJobConcurrency(handlerName),
prefix: this.jobRedisPrefix,
- worker.on('error', err => {
- logger.error('Error in job queue %s.', handlerName, { err })
- })
+ worker.on('error', err => { logger.error('Error in job worker %s.', handlerName, { err }) })
this.workers[handlerName] = worker
}
private buildQueue (handlerName: JobType) {
const queueOptions: QueueOptions = {
this.workers[handlerName] = worker
}
private buildQueue (handlerName: JobType) {
const queueOptions: QueueOptions = {
- this.queues[handlerName] = new Queue(handlerName, queueOptions)
+ const queue = new Queue(handlerName, queueOptions)
+ queue.on('error', err => { logger.error('Error in job queue %s.', handlerName, { err }) })
+
+ this.queues[handlerName] = queue
- this.queueSchedulers[handlerName] = new QueueScheduler(handlerName, queueSchedulerOptions)
+
+ const queueScheduler = new QueueScheduler(handlerName, queueSchedulerOptions)
+ queueScheduler.on('error', err => { logger.error('Error in job queue scheduler %s.', handlerName, { err }) })
+
+ this.queueSchedulers[handlerName] = queueScheduler
- private getRedisConnection () {
- return {
- password: CONFIG.REDIS.AUTH,
- db: CONFIG.REDIS.DB,
- host: CONFIG.REDIS.HOSTNAME,
- port: CONFIG.REDIS.PORT,
- path: CONFIG.REDIS.SOCKET
- }
+ const queueEvents = new QueueEvents(handlerName, queueEventsOptions)
+ queueEvents.on('error', err => { logger.error('Error in job queue events %s.', handlerName, { err }) })
+
+ this.queueEvents[handlerName] = queueEvents
+ start () {
+ const promises = Object.keys(this.workers)
+ .map(handlerName => {
+ const worker: Worker = this.workers[handlerName]
+ const queueScheduler: QueueScheduler = this.queueSchedulers[handlerName]
+ const queueEvent: QueueEvents = this.queueEvents[handlerName]
+
+ return Promise.all([
+ worker.run(),
+ queueScheduler.run(),
+ queueEvent.run()
+ ])
+ })
+
+ return Promise.all(promises)
+ }
+
async pause () {
for (const handlerName of Object.keys(this.workers)) {
const worker: Worker = this.workers[handlerName]
async pause () {
for (const handlerName of Object.keys(this.workers)) {
const worker: Worker = this.workers[handlerName]
for (const handlerName of Object.keys(this.workers)) {
const worker: Worker = this.workers[handlerName]
for (const handlerName of Object.keys(this.workers)) {
const worker: Worker = this.workers[handlerName]
const queue: Queue = this.queues[options.type]
if (queue === undefined) {
logger.error('Unknown queue %s: cannot create job.', options.type)
const queue: Queue = this.queues[options.type]
if (queue === undefined) {
logger.error('Unknown queue %s: cannot create job.', options.type)
return this.flowProducer.add({
...this.buildJobFlowOption(parent),
return this.flowProducer.add({
...this.buildJobFlowOption(parent),
backoff: { delay: 60 * 1000, type: 'exponential' },
attempts: JOB_ATTEMPTS[type],
priority: options.priority,
backoff: { delay: 60 * 1000, type: 'exponential' },
attempts: JOB_ATTEMPTS[type],
priority: options.priority,
- await queue.clean(JOB_COMPLETED_LIFETIME, 100, 'completed')
+ await queue.clean(parseDurationToMs('7 days'), 1000, 'completed')
+ await queue.clean(parseDurationToMs('7 days'), 1000, 'failed')
}
}
private addRepeatableJobs () {
this.queues['videos-views-stats'].add('job', {}, {
}
}
private addRepeatableJobs () {
this.queues['videos-views-stats'].add('job', {}, {
}).catch(err => logger.error('Cannot add repeatable job.', { err }))
if (CONFIG.FEDERATION.VIDEOS.CLEANUP_REMOTE_INTERACTIONS) {
this.queues['activitypub-cleaner'].add('job', {}, {
}).catch(err => logger.error('Cannot add repeatable job.', { err }))
if (CONFIG.FEDERATION.VIDEOS.CLEANUP_REMOTE_INTERACTIONS) {
this.queues['activitypub-cleaner'].add('job', {}, {
+ private buildJobRemovalOptions (queueName: string) {
+ return {
+ removeOnComplete: {
+ // Wants seconds
+ age: (JOB_REMOVAL_OPTIONS.SUCCESS[queueName] || JOB_REMOVAL_OPTIONS.SUCCESS.DEFAULT) / 1000,
+
+ count: JOB_REMOVAL_OPTIONS.COUNT
+ },
+ removeOnFail: {
+ // Wants seconds
+ age: (JOB_REMOVAL_OPTIONS.FAILURE[queueName] || JOB_REMOVAL_OPTIONS.FAILURE.DEFAULT) / 1000,
+
+ count: JOB_REMOVAL_OPTIONS.COUNT / 1000
+ }
+ }
+ }
+