aboutsummaryrefslogtreecommitdiffhomepage
path: root/shared/server-commands/server/jobs.ts
blob: ff309806351c9e088b305236b447246bc6818f11 (plain) (blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
import { expect } from 'chai'
import { wait } from '@shared/core-utils'
import { JobState, JobType, RunnerJobState } from '../../models'
import { PeerTubeServer } from './server'

async function waitJobs (
  serversArg: PeerTubeServer[] | PeerTubeServer,
  options: {
    skipDelayed?: boolean // default false
    runnerJobs?: boolean // default false
  } = {}
) {
  const { skipDelayed = false, runnerJobs = false } = options

  const pendingJobWait = process.env.NODE_PENDING_JOB_WAIT
    ? parseInt(process.env.NODE_PENDING_JOB_WAIT, 10)
    : 250

  let servers: PeerTubeServer[]

  if (Array.isArray(serversArg) === false) servers = [ serversArg as PeerTubeServer ]
  else servers = serversArg as PeerTubeServer[]

  const states: JobState[] = [ 'waiting', 'active' ]
  if (!skipDelayed) states.push('delayed')

  const repeatableJobs: JobType[] = [ 'videos-views-stats', 'activitypub-cleaner' ]
  let pendingRequests: boolean

  function tasksBuilder () {
    const tasks: Promise<any>[] = []

    // Check if each server has pending request
    for (const server of servers) {
      for (const state of states) {

        const jobPromise = server.jobs.list({
          state,
          start: 0,
          count: 10,
          sort: '-createdAt'
        }).then(body => body.data)
          .then(jobs => jobs.filter(j => !repeatableJobs.includes(j.type)))
          .then(jobs => {
            if (jobs.length !== 0) {
              pendingRequests = true
            }
          })

        tasks.push(jobPromise)
      }

      const debugPromise = server.debug.getDebug()
        .then(obj => {
          if (obj.activityPubMessagesWaiting !== 0) {
            pendingRequests = true
          }
        })
      tasks.push(debugPromise)

      if (runnerJobs) {
        const runnerJobsPromise = server.runnerJobs.list({ count: 100 })
          .then(({ data }) => {
            for (const job of data) {
              if (job.state.id !== RunnerJobState.COMPLETED) {
                pendingRequests = true
              }
            }
          })
        tasks.push(runnerJobsPromise)
      }

    }

    return tasks
  }

  do {
    pendingRequests = false
    await Promise.all(tasksBuilder())

    // Retry, in case of new jobs were created
    if (pendingRequests === false) {
      await wait(pendingJobWait)
      await Promise.all(tasksBuilder())
    }

    if (pendingRequests) {
      await wait(pendingJobWait)
    }
  } while (pendingRequests)
}

async function expectNoFailedTranscodingJob (server: PeerTubeServer) {
  const { data } = await server.jobs.listFailed({ jobType: 'video-transcoding' })
  expect(data).to.have.lengthOf(0)
}

// ---------------------------------------------------------------------------

export {
  waitJobs,
  expectNoFailedTranscodingJob
}