aboutsummaryrefslogtreecommitdiffhomepage
diff options
context:
space:
mode:
-rw-r--r--server/initializers/constants.ts22
-rw-r--r--server/lib/job-queue/job-queue.ts46
-rw-r--r--server/lib/schedulers/remove-old-jobs-scheduler.ts2
3 files changed, 61 insertions, 9 deletions
diff --git a/server/initializers/constants.ts b/server/initializers/constants.ts
index 991fe3e85..c8fa8fa2c 100644
--- a/server/initializers/constants.ts
+++ b/server/initializers/constants.ts
@@ -239,7 +239,23 @@ const REQUEST_TIMEOUTS = {
239 REDUNDANCY: JOB_TTL['video-redundancy'] 239 REDUNDANCY: JOB_TTL['video-redundancy']
240} 240}
241 241
242const JOB_COMPLETED_LIFETIME = 60000 * 60 * 24 * 2 // 2 days 242const JOB_REMOVAL_OPTIONS = {
243 COUNT: 10000, // Max jobs to store
244
245 SUCCESS: { // Success jobs
246 'DEFAULT': parseDurationToMs('2 days'),
247
248 'activitypub-http-broadcast-parallel': parseDurationToMs('10 minutes'),
249 'activitypub-http-unicast': parseDurationToMs('1 hour'),
250 'videos-views-stats': parseDurationToMs('3 hours'),
251 'activitypub-refresher': parseDurationToMs('10 hours')
252 },
253
254 FAILURE: { // Failed job
255 DEFAULT: parseDurationToMs('7 days')
256 }
257}
258
243const VIDEO_IMPORT_TIMEOUT = Math.floor(JOB_TTL['video-import'] * 0.9) 259const VIDEO_IMPORT_TIMEOUT = Math.floor(JOB_TTL['video-import'] * 0.9)
244 260
245const SCHEDULER_INTERVALS_MS = { 261const SCHEDULER_INTERVALS_MS = {
@@ -938,6 +954,8 @@ if (process.env.PRODUCTION_CONSTANTS !== 'true') {
938 OVERVIEWS.VIDEOS.SAMPLE_THRESHOLD = 2 954 OVERVIEWS.VIDEOS.SAMPLE_THRESHOLD = 2
939 955
940 PLUGIN_EXTERNAL_AUTH_TOKEN_LIFETIME = 5000 956 PLUGIN_EXTERNAL_AUTH_TOKEN_LIFETIME = 5000
957
958 JOB_REMOVAL_OPTIONS.SUCCESS['videos-views-stats'] = 10000
941 } 959 }
942 960
943 if (isTestInstance()) { 961 if (isTestInstance()) {
@@ -1069,7 +1087,7 @@ export {
1069 CRAWL_REQUEST_CONCURRENCY, 1087 CRAWL_REQUEST_CONCURRENCY,
1070 DEFAULT_AUDIO_RESOLUTION, 1088 DEFAULT_AUDIO_RESOLUTION,
1071 BINARY_CONTENT_TYPES, 1089 BINARY_CONTENT_TYPES,
1072 JOB_COMPLETED_LIFETIME, 1090 JOB_REMOVAL_OPTIONS,
1073 HTTP_SIGNATURE, 1091 HTTP_SIGNATURE,
1074 VIDEO_IMPORT_STATES, 1092 VIDEO_IMPORT_STATES,
1075 VIDEO_CHANNEL_SYNC_STATE, 1093 VIDEO_CHANNEL_SYNC_STATE,
diff --git a/server/lib/job-queue/job-queue.ts b/server/lib/job-queue/job-queue.ts
index 6bc59732f..91366c481 100644
--- a/server/lib/job-queue/job-queue.ts
+++ b/server/lib/job-queue/job-queue.ts
@@ -41,8 +41,16 @@ import {
41 VideoTranscodingPayload 41 VideoTranscodingPayload
42} from '../../../shared/models' 42} from '../../../shared/models'
43import { logger } from '../../helpers/logger' 43import { logger } from '../../helpers/logger'
44import { JOB_ATTEMPTS, JOB_COMPLETED_LIFETIME, JOB_CONCURRENCY, JOB_TTL, REPEAT_JOBS, WEBSERVER } from '../../initializers/constants' 44import {
45 JOB_ATTEMPTS,
46 JOB_CONCURRENCY,
47 JOB_REMOVAL_OPTIONS,
48 JOB_TTL,
49 REPEAT_JOBS,
50 WEBSERVER
51} from '../../initializers/constants'
45import { Hooks } from '../plugins/hooks' 52import { Hooks } from '../plugins/hooks'
53import { Redis } from '../redis'
46import { processActivityPubCleaner } from './handlers/activitypub-cleaner' 54import { processActivityPubCleaner } from './handlers/activitypub-cleaner'
47import { processActivityPubFollow } from './handlers/activitypub-follow' 55import { processActivityPubFollow } from './handlers/activitypub-follow'
48import { processActivityPubHttpSequentialBroadcast, processActivityPubParallelHttpBroadcast } from './handlers/activitypub-http-broadcast' 56import { processActivityPubHttpSequentialBroadcast, processActivityPubParallelHttpBroadcast } from './handlers/activitypub-http-broadcast'
@@ -63,7 +71,7 @@ import { processVideoLiveEnding } from './handlers/video-live-ending'
63import { processVideoStudioEdition } from './handlers/video-studio-edition' 71import { processVideoStudioEdition } from './handlers/video-studio-edition'
64import { processVideoTranscoding } from './handlers/video-transcoding' 72import { processVideoTranscoding } from './handlers/video-transcoding'
65import { processVideosViewsStats } from './handlers/video-views-stats' 73import { processVideosViewsStats } from './handlers/video-views-stats'
66import { Redis } from '../redis' 74import { parseDurationToMs } from '@server/helpers/core-utils'
67 75
68export type CreateJobArgument = 76export type CreateJobArgument =
69 { type: 'activitypub-http-broadcast', payload: ActivitypubHttpBroadcastPayload } | 77 { type: 'activitypub-http-broadcast', payload: ActivitypubHttpBroadcastPayload } |
@@ -373,7 +381,7 @@ class JobQueue {
373 }) 381 })
374 } 382 }
375 383
376 private buildJobFlowOption (job: CreateJobArgument & CreateJobOptions) { 384 private buildJobFlowOption (job: CreateJobArgument & CreateJobOptions): FlowJob {
377 return { 385 return {
378 name: 'job', 386 name: 'job',
379 data: job.payload, 387 data: job.payload,
@@ -387,7 +395,9 @@ class JobQueue {
387 backoff: { delay: 60 * 1000, type: 'exponential' }, 395 backoff: { delay: 60 * 1000, type: 'exponential' },
388 attempts: JOB_ATTEMPTS[type], 396 attempts: JOB_ATTEMPTS[type],
389 priority: options.priority, 397 priority: options.priority,
390 delay: options.delay 398 delay: options.delay,
399
400 ...this.buildJobRemovalOptions(type)
391 } 401 }
392 } 402 }
393 403
@@ -482,18 +492,23 @@ class JobQueue {
482 async removeOldJobs () { 492 async removeOldJobs () {
483 for (const key of Object.keys(this.queues)) { 493 for (const key of Object.keys(this.queues)) {
484 const queue: Queue = this.queues[key] 494 const queue: Queue = this.queues[key]
485 await queue.clean(JOB_COMPLETED_LIFETIME, 100, 'completed') 495 await queue.clean(parseDurationToMs('7 days'), 100, 'completed')
496 await queue.clean(parseDurationToMs('7 days'), 100, 'failed')
486 } 497 }
487 } 498 }
488 499
489 private addRepeatableJobs () { 500 private addRepeatableJobs () {
490 this.queues['videos-views-stats'].add('job', {}, { 501 this.queues['videos-views-stats'].add('job', {}, {
491 repeat: REPEAT_JOBS['videos-views-stats'] 502 repeat: REPEAT_JOBS['videos-views-stats'],
503
504 ...this.buildJobRemovalOptions('videos-views-stats')
492 }).catch(err => logger.error('Cannot add repeatable job.', { err })) 505 }).catch(err => logger.error('Cannot add repeatable job.', { err }))
493 506
494 if (CONFIG.FEDERATION.VIDEOS.CLEANUP_REMOTE_INTERACTIONS) { 507 if (CONFIG.FEDERATION.VIDEOS.CLEANUP_REMOTE_INTERACTIONS) {
495 this.queues['activitypub-cleaner'].add('job', {}, { 508 this.queues['activitypub-cleaner'].add('job', {}, {
496 repeat: REPEAT_JOBS['activitypub-cleaner'] 509 repeat: REPEAT_JOBS['activitypub-cleaner'],
510
511 ...this.buildJobRemovalOptions('activitypub-cleaner')
497 }).catch(err => logger.error('Cannot add repeatable job.', { err })) 512 }).catch(err => logger.error('Cannot add repeatable job.', { err }))
498 } 513 }
499 } 514 }
@@ -505,6 +520,23 @@ class JobQueue {
505 return JOB_CONCURRENCY[jobType] 520 return JOB_CONCURRENCY[jobType]
506 } 521 }
507 522
523 private buildJobRemovalOptions (queueName: string) {
524 return {
525 removeOnComplete: {
526 // Wants seconds
527 age: (JOB_REMOVAL_OPTIONS.SUCCESS[queueName] || JOB_REMOVAL_OPTIONS.SUCCESS.DEFAULT) / 1000,
528
529 count: JOB_REMOVAL_OPTIONS.COUNT
530 },
531 removeOnFail: {
532 // Wants seconds
533 age: (JOB_REMOVAL_OPTIONS.FAILURE[queueName] || JOB_REMOVAL_OPTIONS.FAILURE.DEFAULT) / 1000,
534
535 count: JOB_REMOVAL_OPTIONS.COUNT / 1000
536 }
537 }
538 }
539
508 static get Instance () { 540 static get Instance () {
509 return this.instance || (this.instance = new this()) 541 return this.instance || (this.instance = new this())
510 } 542 }
diff --git a/server/lib/schedulers/remove-old-jobs-scheduler.ts b/server/lib/schedulers/remove-old-jobs-scheduler.ts
index 879846999..6b05ea144 100644
--- a/server/lib/schedulers/remove-old-jobs-scheduler.ts
+++ b/server/lib/schedulers/remove-old-jobs-scheduler.ts
@@ -4,6 +4,8 @@ import { SCHEDULER_INTERVALS_MS } from '../../initializers/constants'
4import { JobQueue } from '../job-queue' 4import { JobQueue } from '../job-queue'
5import { AbstractScheduler } from './abstract-scheduler' 5import { AbstractScheduler } from './abstract-scheduler'
6 6
7// FIXME: delete this scheduler in a few versions (introduced in 5.0)
8// We introduced job removal directly using bullmq option but we still need to delete old jobs
7export class RemoveOldJobsScheduler extends AbstractScheduler { 9export class RemoveOldJobsScheduler extends AbstractScheduler {
8 10
9 private static instance: AbstractScheduler 11 private static instance: AbstractScheduler