Move job queue to redis

We'll use it as cache in the future.

/!\ You'll loose your old jobs (pending jobs too) so upgrade only when
you don't have pending job anymore.

6 files changed, 45 insertions, 37 deletions

diff --git a/server/tests/api/check-params/jobs.ts b/server/tests/api/check-params/jobs.ts
index b12818bb1..ce3ac8809 100644
--- a/server/tests/api/check-params/jobs.ts
+++ b/server/tests/api/check-params/jobs.ts
@@ -7,7 +7,7 @@ import { checkBadCountPagination, checkBadSortPagination, checkBadStartPaginatio
 import { makeGetRequest } from '../../utils/requests/requests'
 
 describe('Test jobs API validators', function () {
-  const path = '/api/v1/jobs/'
+  const path = '/api/v1/jobs/failed'
   let server: ServerInfo
   let userAccessToken = ''
 
@@ -31,6 +31,15 @@ describe('Test jobs API validators', function () {
   })
 
   describe('When listing jobs', function () {
+
+    it('Should fail with a bad state', async function () {
+      await makeGetRequest({
+        url: server.url,
+        token: server.accessToken,
+        path: path + 'ade'
+      })
+    })
+
     it('Should fail with a bad start pagination', async function () {
       await checkBadStartPagination(server.url, path, server.accessToken)
     })
diff --git a/server/tests/api/server/handle-down.ts b/server/tests/api/server/handle-down.ts
index de4e77b2f..4cedeb89e 100644
--- a/server/tests/api/server/handle-down.ts
+++ b/server/tests/api/server/handle-down.ts
@@ -2,6 +2,7 @@
 
 import * as chai from 'chai'
 import 'mocha'
+import { JobState } from '../../../../shared/models'
 import { VideoPrivacy } from '../../../../shared/models/videos'
 import { VideoCommentThreadTree } from '../../../../shared/models/videos/video-comment.model'
 import { completeVideoCheck, getVideo, immutableAssign, reRunServer, viewVideo } from '../../utils'
@@ -139,12 +140,11 @@ describe('Test handle downs', function () {
   })
 
   it('Should not have pending/processing jobs anymore', async function () {
-    const res = await getJobsListPaginationAndSort(servers[0].url, servers[0].accessToken, 0, 50, '-createdAt')
-    const jobs = res.body.data
+    const states: JobState[] = [ 'inactive', 'active' ]
 
-    for (const job of jobs) {
-      expect(job.state).not.to.equal('pending')
-      expect(job.state).not.to.equal('processing')
+    for (const state of states) {
+      const res = await getJobsListPaginationAndSort(servers[ 0 ].url, servers[ 0 ].accessToken, state,0, 50, '-createdAt')
+      expect(res.body.data).to.have.length(0)
     }
   })
 
diff --git a/server/tests/api/server/jobs.ts b/server/tests/api/server/jobs.ts
index 2e17e71a4..671498769 100644
--- a/server/tests/api/server/jobs.ts
+++ b/server/tests/api/server/jobs.ts
@@ -35,20 +35,20 @@ describe('Test jobs', function () {
   })
 
   it('Should list jobs', async function () {
-    const res = await getJobsList(servers[1].url, servers[1].accessToken)
+    const res = await getJobsList(servers[1].url, servers[1].accessToken, 'complete')
     expect(res.body.total).to.be.above(2)
     expect(res.body.data).to.have.length.above(2)
   })
 
   it('Should list jobs with sort and pagination', async function () {
-    const res = await getJobsListPaginationAndSort(servers[1].url, servers[1].accessToken, 4, 1, 'createdAt')
+    const res = await getJobsListPaginationAndSort(servers[1].url, servers[1].accessToken, 'complete', 1, 1, 'createdAt')
     expect(res.body.total).to.be.above(2)
     expect(res.body.data).to.have.lengthOf(1)
 
     const job = res.body.data[0]
-    expect(job.state).to.equal('success')
-    expect(job.category).to.equal('transcoding')
-    expect(job.handlerName).to.have.length.above(3)
+
+    expect(job.state).to.equal('complete')
+    expect(job.type).to.equal('activitypub-http-unicast')
     expect(dateIsValid(job.createdAt)).to.be.true
     expect(dateIsValid(job.updatedAt)).to.be.true
   })
diff --git a/server/tests/api/videos/multiple-servers.ts b/server/tests/api/videos/multiple-servers.ts
index 4c4b5123d..0215b3011 100644
--- a/server/tests/api/videos/multiple-servers.ts
+++ b/server/tests/api/videos/multiple-servers.ts
@@ -475,16 +475,17 @@ describe('Test multiple servers', function () {
     it('Should like and dislikes videos on different services', async function () {
       this.timeout(20000)
 
-      const tasks: Promise<any>[] = []
-      tasks.push(rateVideo(servers[0].url, servers[0].accessToken, remoteVideosServer1[0], 'like'))
-      tasks.push(rateVideo(servers[0].url, servers[0].accessToken, remoteVideosServer1[0], 'dislike'))
-      tasks.push(rateVideo(servers[0].url, servers[0].accessToken, remoteVideosServer1[0], 'like'))
-      tasks.push(rateVideo(servers[2].url, servers[2].accessToken, localVideosServer3[1], 'like'))
-      tasks.push(rateVideo(servers[2].url, servers[2].accessToken, localVideosServer3[1], 'dislike'))
-      tasks.push(rateVideo(servers[2].url, servers[2].accessToken, remoteVideosServer3[1], 'dislike'))
-      tasks.push(rateVideo(servers[2].url, servers[2].accessToken, remoteVideosServer3[0], 'like'))
-
-      await Promise.all(tasks)
+      await rateVideo(servers[0].url, servers[0].accessToken, remoteVideosServer1[0], 'like')
+      await wait(200)
+      await rateVideo(servers[0].url, servers[0].accessToken, remoteVideosServer1[0], 'dislike')
+      await wait(200)
+      await rateVideo(servers[0].url, servers[0].accessToken, remoteVideosServer1[0], 'like')
+      await rateVideo(servers[2].url, servers[2].accessToken, localVideosServer3[1], 'like')
+      await wait(200)
+      await rateVideo(servers[2].url, servers[2].accessToken, localVideosServer3[1], 'dislike')
+      await rateVideo(servers[2].url, servers[2].accessToken, remoteVideosServer3[1], 'dislike')
+      await wait(200)
+      await rateVideo(servers[2].url, servers[2].accessToken, remoteVideosServer3[0], 'like')
 
       await wait(10000)
 
diff --git a/server/tests/real-world/real-world.ts b/server/tests/real-world/real-world.ts
index e41203351..7f67525ed 100644
--- a/server/tests/real-world/real-world.ts
+++ b/server/tests/real-world/real-world.ts
@@ -3,6 +3,7 @@ process.env.NODE_ENV = 'test'
 
 import * as program from 'commander'
 import { Video, VideoFile, VideoRateType } from '../../../shared'
+import { JobState } from '../../../shared/models'
 import {
   flushAndRunMultipleServers,
   flushTests, follow,
@@ -346,23 +347,19 @@ function goodbye () {
 }
 
 async function isTherePendingRequests (servers: ServerInfo[]) {
+  const states: JobState[] = [ 'inactive', 'active' ]
   const tasks: Promise<any>[] = []
   let pendingRequests = false
 
   // Check if each server has pending request
   for (const server of servers) {
-    const p = getJobsListPaginationAndSort(server.url, server.accessToken, 0, 10, '-createdAt')
-      .then(res => {
-        const jobs = res.body.data
-
-        for (const job of jobs) {
-          if (job.state === 'pending' || job.state === 'processing') {
-            pendingRequests = true
-          }
-        }
-      })
-
-    tasks.push(p)
+    for (const state of states) {
+      const p = getJobsListPaginationAndSort(server.url, server.accessToken, state, 0, 10, '-createdAt')
+        .then(res => {
+          if (res.body.total > 0) pendingRequests = true
+        })
+      tasks.push(p)
+    }
   }
 
   await Promise.all(tasks)
diff --git a/server/tests/utils/server/jobs.ts b/server/tests/utils/server/jobs.ts
index 0a8c51575..4053dd40b 100644
--- a/server/tests/utils/server/jobs.ts
+++ b/server/tests/utils/server/jobs.ts
@@ -1,7 +1,8 @@
 import * as request from 'supertest'
+import { JobState } from '../../../../shared/models'
 
-function getJobsList (url: string, accessToken: string) {
-  const path = '/api/v1/jobs'
+function getJobsList (url: string, accessToken: string, state: JobState) {
+  const path = '/api/v1/jobs/' + state
 
   return request(url)
           .get(path)
@@ -11,8 +12,8 @@ function getJobsList (url: string, accessToken: string) {
           .expect('Content-Type', /json/)
 }
 
-function getJobsListPaginationAndSort (url: string, accessToken: string, start: number, count: number, sort: string) {
-  const path = '/api/v1/jobs'
+function getJobsListPaginationAndSort (url: string, accessToken: string, state: JobState, start: number, count: number, sort: string) {
+  const path = '/api/v1/jobs/' + state
 
   return request(url)
           .get(path)