shared/server-commands/server/jobs.ts

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