aboutsummaryrefslogtreecommitdiffhomepage
path: root/server/lib/job-queue
diff options
context:
space:
mode:
Diffstat (limited to 'server/lib/job-queue')
-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
4 files changed, 27 insertions, 11 deletions
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