From 71e3dfda4e2bcc228415c0d66b09a84bb73dddd1 Mon Sep 17 00:00:00 2001 From: Chocobozzz Date: Wed, 9 May 2018 09:08:22 +0200 Subject: Add timeout and TTL to request jobs --- server/initializers/constants.ts | 9 +++++--- .../handlers/activitypub-http-broadcast.ts | 5 +++-- .../job-queue/handlers/activitypub-http-fetcher.ts | 5 +++-- .../job-queue/handlers/activitypub-http-unicast.ts | 4 +++- server/lib/job-queue/job-queue.ts | 24 ++++++++++++++++------ 5 files changed, 33 insertions(+), 14 deletions(-) (limited to 'server') diff --git a/server/initializers/constants.ts b/server/initializers/constants.ts index 365b8617d..6556aa168 100644 --- a/server/initializers/constants.ts +++ b/server/initializers/constants.ts @@ -78,9 +78,10 @@ const JOB_CONCURRENCY: { [ id in JobType ]: number } = { 'video-file': 1, 'email': 5 } -const BROADCAST_CONCURRENCY = 5 // How many requests in parallel we do in activitypub-http-broadcast job -// 2 days -const JOB_COMPLETED_LIFETIME = 60000 * 60 * 24 * 2 +const BROADCAST_CONCURRENCY = 10 // How many requests in parallel we do in activitypub-http-broadcast job +const JOB_REQUEST_TIMEOUT = 3000 // 3 seconds +const JOB_REQUEST_TTL = 60000 * 10 // 10 minutes +const JOB_COMPLETED_LIFETIME = 60000 * 60 * 24 * 2 // 2 days // 1 hour let SCHEDULER_INTERVAL = 60000 * 60 @@ -466,6 +467,8 @@ export { VIDEO_RATE_TYPES, VIDEO_MIMETYPE_EXT, VIDEO_TRANSCODING_FPS, + JOB_REQUEST_TIMEOUT, + JOB_REQUEST_TTL, USER_PASSWORD_RESET_LIFETIME, IMAGE_MIMETYPE_EXT, SCHEDULER_INTERVAL, diff --git a/server/lib/job-queue/handlers/activitypub-http-broadcast.ts b/server/lib/job-queue/handlers/activitypub-http-broadcast.ts index 38b8393f4..d8b8ec222 100644 --- a/server/lib/job-queue/handlers/activitypub-http-broadcast.ts +++ b/server/lib/job-queue/handlers/activitypub-http-broadcast.ts @@ -4,7 +4,7 @@ import { logger } from '../../../helpers/logger' import { doRequest } from '../../../helpers/requests' import { ActorFollowModel } from '../../../models/activitypub/actor-follow' import { buildSignedRequestOptions, computeBody } from './utils/activitypub-http-utils' -import { BROADCAST_CONCURRENCY } from '../../../initializers' +import { BROADCAST_CONCURRENCY, JOB_REQUEST_TIMEOUT } from '../../../initializers' export type ActivitypubHttpBroadcastPayload = { uris: string[] @@ -24,7 +24,8 @@ async function processActivityPubHttpBroadcast (job: kue.Job) { method: 'POST', uri: '', json: body, - httpSignature: httpSignatureOptions + httpSignature: httpSignatureOptions, + timeout: JOB_REQUEST_TIMEOUT } const badUrls: string[] = [] diff --git a/server/lib/job-queue/handlers/activitypub-http-fetcher.ts b/server/lib/job-queue/handlers/activitypub-http-fetcher.ts index 062211c85..4683beb2f 100644 --- a/server/lib/job-queue/handlers/activitypub-http-fetcher.ts +++ b/server/lib/job-queue/handlers/activitypub-http-fetcher.ts @@ -1,7 +1,7 @@ import * as kue from 'kue' import { logger } from '../../../helpers/logger' import { doRequest } from '../../../helpers/requests' -import { ACTIVITY_PUB } from '../../../initializers' +import { ACTIVITY_PUB, JOB_REQUEST_TIMEOUT } from '../../../initializers' import { processActivities } from '../../activitypub/process' import { ActivitypubHttpBroadcastPayload } from './activitypub-http-broadcast' @@ -18,7 +18,8 @@ async function processActivityPubHttpFetcher (job: kue.Job) { method: 'GET', uri: '', json: true, - activityPub: true + activityPub: true, + timeout: JOB_REQUEST_TIMEOUT } for (const uri of payload.uris) { diff --git a/server/lib/job-queue/handlers/activitypub-http-unicast.ts b/server/lib/job-queue/handlers/activitypub-http-unicast.ts index e1e1824e5..173f3bb52 100644 --- a/server/lib/job-queue/handlers/activitypub-http-unicast.ts +++ b/server/lib/job-queue/handlers/activitypub-http-unicast.ts @@ -3,6 +3,7 @@ import { logger } from '../../../helpers/logger' import { doRequest } from '../../../helpers/requests' import { ActorFollowModel } from '../../../models/activitypub/actor-follow' import { buildSignedRequestOptions, computeBody } from './utils/activitypub-http-utils' +import { JOB_REQUEST_TIMEOUT } from '../../../initializers' export type ActivitypubHttpUnicastPayload = { uri: string @@ -23,7 +24,8 @@ async function processActivityPubHttpUnicast (job: kue.Job) { method: 'POST', uri, json: body, - httpSignature: httpSignatureOptions + httpSignature: httpSignatureOptions, + timeout: JOB_REQUEST_TIMEOUT } try { diff --git a/server/lib/job-queue/job-queue.ts b/server/lib/job-queue/job-queue.ts index bf40a9206..acc69ef24 100644 --- a/server/lib/job-queue/job-queue.ts +++ b/server/lib/job-queue/job-queue.ts @@ -1,7 +1,7 @@ import * as kue from 'kue' import { JobState, JobType } from '../../../shared/models' import { logger } from '../../helpers/logger' -import { CONFIG, JOB_ATTEMPTS, JOB_COMPLETED_LIFETIME, JOB_CONCURRENCY } from '../../initializers' +import { CONFIG, JOB_ATTEMPTS, JOB_COMPLETED_LIFETIME, JOB_CONCURRENCY, JOB_REQUEST_TTL } from '../../initializers' import { Redis } from '../redis' import { ActivitypubHttpBroadcastPayload, processActivityPubHttpBroadcast } from './handlers/activitypub-http-broadcast' import { ActivitypubHttpFetcherPayload, processActivityPubHttpFetcher } from './handlers/activitypub-http-fetcher' @@ -27,6 +27,13 @@ const handlers: { [ id in JobType ]: (job: kue.Job) => Promise} = { 'email': processEmail } +const jobsWithTLL: JobType[] = [ + 'activitypub-http-broadcast', + 'activitypub-http-unicast', + 'activitypub-http-fetcher', + 'activitypub-follow' +] + class JobQueue { private static instance: JobQueue @@ -77,16 +84,21 @@ class JobQueue { createJob (obj: CreateJobArgument, priority = 'normal') { return new Promise((res, rej) => { - this.jobQueue + let job = this.jobQueue .create(obj.type, obj.payload) .priority(priority) .attempts(JOB_ATTEMPTS[obj.type]) .backoff({ delay: 60 * 1000, type: 'exponential' }) - .save(err => { - if (err) return rej(err) - return res() - }) + if (jobsWithTLL.indexOf(obj.type) !== -1) { + job = job.ttl(JOB_REQUEST_TTL) + } + + return job.save(err => { + if (err) return rej(err) + + return res() + }) }) } -- cgit v1.2.3