shared/server-commands/server/jobs.ts

   1
   2 import { expect } from 'chai'
   3 import { JobState, JobType } from '../../models'
   4 import { wait } from '../miscs'
   5 import { PeerTubeServer } from './server'
   6
   7 async function waitJobs (serversArg: PeerTubeServer[] | PeerTubeServer, skipDelayed = false) {
   8   const pendingJobWait = process.env.NODE_PENDING_JOB_WAIT
   9     ? parseInt(process.env.NODE_PENDING_JOB_WAIT, 10)
  10     : 250
  11
  12   let servers: PeerTubeServer[]
  13
  14   if (Array.isArray(serversArg) === false) servers = [ serversArg as PeerTubeServer ]
  15   else servers = serversArg as PeerTubeServer[]
  16
  17   const states: JobState[] = [ 'waiting', 'active' ]
  18   if (!skipDelayed) states.push('delayed')
  19
  20   const repeatableJobs: JobType[] = [ 'videos-views-stats', 'activitypub-cleaner' ]
  21   let pendingRequests: boolean
  22
  23   function tasksBuilder () {
  24     const tasks: Promise<any>[] = []
  25
  26     // Check if each server has pending request
  27     for (const server of servers) {
  28       for (const state of states) {
  29         const p = server.jobs.list({
  30           state,
  31           start: 0,
  32           count: 10,
  33           sort: '-createdAt'
  34         }).then(body => body.data)
  35           .then(jobs => jobs.filter(j => !repeatableJobs.includes(j.type)))
  36           .then(jobs => {
  37             if (jobs.length !== 0) {
  38               pendingRequests = true
  39             }
  40           })
  41
  42         tasks.push(p)
  43       }
  44
  45       const p = server.debug.getDebug()
  46         .then(obj => {
  47           if (obj.activityPubMessagesWaiting !== 0) {
  48             pendingRequests = true
  49           }
  50         })
  51
  52       tasks.push(p)
  53     }
  54
  55     return tasks
  56   }
  57
  58   do {
  59     pendingRequests = false
  60     await Promise.all(tasksBuilder())
  61
  62     // Retry, in case of new jobs were created
  63     if (pendingRequests === false) {
  64       await wait(pendingJobWait)
  65       await Promise.all(tasksBuilder())
  66     }
  67
  68     if (pendingRequests) {
  69       await wait(pendingJobWait)
  70     }
  71   } while (pendingRequests)
  72 }
  73
  74 async function expectNoFailedTranscodingJob (server: PeerTubeServer) {
  75   const { data } = await server.jobs.listFailed({ jobType: 'video-transcoding' })
  76   expect(data).to.have.lengthOf(0)
  77 }
  78
  79 // ---------------------------------------------------------------------------
  80
  81 export {
  82   waitJobs,
  83   expectNoFailedTranscodingJob
  84 }