shared/server-commands/server/jobs.ts

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