aboutsummaryrefslogtreecommitdiffhomepage
path: root/server
diff options
context:
space:
mode:
authorChocobozzz <me@florianbigard.com>2018-05-09 09:08:22 +0200
committerChocobozzz <me@florianbigard.com>2018-05-09 09:08:22 +0200
commit71e3dfda4e2bcc228415c0d66b09a84bb73dddd1 (patch)
treef7b17a3f01b5e796b73debcd3f11fa0bb56f18bb /server
parent7797350a0e539b872bdcb59aa07fa9b16cb2db7a (diff)
downloadPeerTube-71e3dfda4e2bcc228415c0d66b09a84bb73dddd1.tar.gz
PeerTube-71e3dfda4e2bcc228415c0d66b09a84bb73dddd1.tar.zst
PeerTube-71e3dfda4e2bcc228415c0d66b09a84bb73dddd1.zip
Add timeout and TTL to request jobs
Diffstat (limited to 'server')
-rw-r--r--server/initializers/constants.ts9
-rw-r--r--server/lib/job-queue/handlers/activitypub-http-broadcast.ts5
-rw-r--r--server/lib/job-queue/handlers/activitypub-http-fetcher.ts5
-rw-r--r--server/lib/job-queue/handlers/activitypub-http-unicast.ts4
-rw-r--r--server/lib/job-queue/job-queue.ts24
5 files changed, 33 insertions, 14 deletions
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 } = {
78 'video-file': 1, 78 'video-file': 1,
79 'email': 5 79 'email': 5
80} 80}
81const BROADCAST_CONCURRENCY = 5 // How many requests in parallel we do in activitypub-http-broadcast job 81const BROADCAST_CONCURRENCY = 10 // How many requests in parallel we do in activitypub-http-broadcast job
82// 2 days 82const JOB_REQUEST_TIMEOUT = 3000 // 3 seconds
83const JOB_COMPLETED_LIFETIME = 60000 * 60 * 24 * 2 83const JOB_REQUEST_TTL = 60000 * 10 // 10 minutes
84const JOB_COMPLETED_LIFETIME = 60000 * 60 * 24 * 2 // 2 days
84 85
85// 1 hour 86// 1 hour
86let SCHEDULER_INTERVAL = 60000 * 60 87let SCHEDULER_INTERVAL = 60000 * 60
@@ -466,6 +467,8 @@ export {
466 VIDEO_RATE_TYPES, 467 VIDEO_RATE_TYPES,
467 VIDEO_MIMETYPE_EXT, 468 VIDEO_MIMETYPE_EXT,
468 VIDEO_TRANSCODING_FPS, 469 VIDEO_TRANSCODING_FPS,
470 JOB_REQUEST_TIMEOUT,
471 JOB_REQUEST_TTL,
469 USER_PASSWORD_RESET_LIFETIME, 472 USER_PASSWORD_RESET_LIFETIME,
470 IMAGE_MIMETYPE_EXT, 473 IMAGE_MIMETYPE_EXT,
471 SCHEDULER_INTERVAL, 474 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'
4import { doRequest } from '../../../helpers/requests' 4import { doRequest } from '../../../helpers/requests'
5import { ActorFollowModel } from '../../../models/activitypub/actor-follow' 5import { ActorFollowModel } from '../../../models/activitypub/actor-follow'
6import { buildSignedRequestOptions, computeBody } from './utils/activitypub-http-utils' 6import { buildSignedRequestOptions, computeBody } from './utils/activitypub-http-utils'
7import { BROADCAST_CONCURRENCY } from '../../../initializers' 7import { BROADCAST_CONCURRENCY, JOB_REQUEST_TIMEOUT } from '../../../initializers'
8 8
9export type ActivitypubHttpBroadcastPayload = { 9export type ActivitypubHttpBroadcastPayload = {
10 uris: string[] 10 uris: string[]
@@ -24,7 +24,8 @@ async function processActivityPubHttpBroadcast (job: kue.Job) {
24 method: 'POST', 24 method: 'POST',
25 uri: '', 25 uri: '',
26 json: body, 26 json: body,
27 httpSignature: httpSignatureOptions 27 httpSignature: httpSignatureOptions,
28 timeout: JOB_REQUEST_TIMEOUT
28 } 29 }
29 30
30 const badUrls: string[] = [] 31 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 @@
1import * as kue from 'kue' 1import * as kue from 'kue'
2import { logger } from '../../../helpers/logger' 2import { logger } from '../../../helpers/logger'
3import { doRequest } from '../../../helpers/requests' 3import { doRequest } from '../../../helpers/requests'
4import { ACTIVITY_PUB } from '../../../initializers' 4import { ACTIVITY_PUB, JOB_REQUEST_TIMEOUT } from '../../../initializers'
5import { processActivities } from '../../activitypub/process' 5import { processActivities } from '../../activitypub/process'
6import { ActivitypubHttpBroadcastPayload } from './activitypub-http-broadcast' 6import { ActivitypubHttpBroadcastPayload } from './activitypub-http-broadcast'
7 7
@@ -18,7 +18,8 @@ async function processActivityPubHttpFetcher (job: kue.Job) {
18 method: 'GET', 18 method: 'GET',
19 uri: '', 19 uri: '',
20 json: true, 20 json: true,
21 activityPub: true 21 activityPub: true,
22 timeout: JOB_REQUEST_TIMEOUT
22 } 23 }
23 24
24 for (const uri of payload.uris) { 25 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'
3import { doRequest } from '../../../helpers/requests' 3import { doRequest } from '../../../helpers/requests'
4import { ActorFollowModel } from '../../../models/activitypub/actor-follow' 4import { ActorFollowModel } from '../../../models/activitypub/actor-follow'
5import { buildSignedRequestOptions, computeBody } from './utils/activitypub-http-utils' 5import { buildSignedRequestOptions, computeBody } from './utils/activitypub-http-utils'
6import { JOB_REQUEST_TIMEOUT } from '../../../initializers'
6 7
7export type ActivitypubHttpUnicastPayload = { 8export type ActivitypubHttpUnicastPayload = {
8 uri: string 9 uri: string
@@ -23,7 +24,8 @@ async function processActivityPubHttpUnicast (job: kue.Job) {
23 method: 'POST', 24 method: 'POST',
24 uri, 25 uri,
25 json: body, 26 json: body,
26 httpSignature: httpSignatureOptions 27 httpSignature: httpSignatureOptions,
28 timeout: JOB_REQUEST_TIMEOUT
27 } 29 }
28 30
29 try { 31 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 @@
1import * as kue from 'kue' 1import * as kue from 'kue'
2import { JobState, JobType } from '../../../shared/models' 2import { JobState, JobType } from '../../../shared/models'
3import { logger } from '../../helpers/logger' 3import { logger } from '../../helpers/logger'
4import { CONFIG, JOB_ATTEMPTS, JOB_COMPLETED_LIFETIME, JOB_CONCURRENCY } from '../../initializers' 4import { CONFIG, JOB_ATTEMPTS, JOB_COMPLETED_LIFETIME, JOB_CONCURRENCY, JOB_REQUEST_TTL } from '../../initializers'
5import { Redis } from '../redis' 5import { Redis } from '../redis'
6import { ActivitypubHttpBroadcastPayload, processActivityPubHttpBroadcast } from './handlers/activitypub-http-broadcast' 6import { ActivitypubHttpBroadcastPayload, processActivityPubHttpBroadcast } from './handlers/activitypub-http-broadcast'
7import { ActivitypubHttpFetcherPayload, processActivityPubHttpFetcher } from './handlers/activitypub-http-fetcher' 7import { ActivitypubHttpFetcherPayload, processActivityPubHttpFetcher } from './handlers/activitypub-http-fetcher'
@@ -27,6 +27,13 @@ const handlers: { [ id in JobType ]: (job: kue.Job) => Promise<any>} = {
27 'email': processEmail 27 'email': processEmail
28} 28}
29 29
30const jobsWithTLL: JobType[] = [
31 'activitypub-http-broadcast',
32 'activitypub-http-unicast',
33 'activitypub-http-fetcher',
34 'activitypub-follow'
35]
36
30class JobQueue { 37class JobQueue {
31 38
32 private static instance: JobQueue 39 private static instance: JobQueue
@@ -77,16 +84,21 @@ class JobQueue {
77 84
78 createJob (obj: CreateJobArgument, priority = 'normal') { 85 createJob (obj: CreateJobArgument, priority = 'normal') {
79 return new Promise((res, rej) => { 86 return new Promise((res, rej) => {
80 this.jobQueue 87 let job = this.jobQueue
81 .create(obj.type, obj.payload) 88 .create(obj.type, obj.payload)
82 .priority(priority) 89 .priority(priority)
83 .attempts(JOB_ATTEMPTS[obj.type]) 90 .attempts(JOB_ATTEMPTS[obj.type])
84 .backoff({ delay: 60 * 1000, type: 'exponential' }) 91 .backoff({ delay: 60 * 1000, type: 'exponential' })
85 .save(err => {
86 if (err) return rej(err)
87 92
88 return res() 93 if (jobsWithTLL.indexOf(obj.type) !== -1) {
89 }) 94 job = job.ttl(JOB_REQUEST_TTL)
95 }
96
97 return job.save(err => {
98 if (err) return rej(err)
99
100 return res()
101 })
90 }) 102 })
91 } 103 }
92 104