Add timeout and TTL to request jobs

1 files changed, 18 insertions, 6 deletions

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<any>} = {
   '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()
+      })
     })
   }