Fix peertube runner concurrency

[github/Chocobozzz/PeerTube.git] / packages / peertube-runner / server / server.ts

diff --git a/packages/peertube-runner/server/server.ts b/packages/peertube-runner/server/server.ts
index e851dfc7cc743521a217c357c101760949dd4394..cb1533dc6db6095bab2bdea88acd4ea5afda1ddc 100644 (file)
--- a/packages/peertube-runner/server/server.ts
+++ b/packages/peertube-runner/server/server.ts
@@ -8,6 +8,7 @@ import { ConfigManager } from '../shared'
 import { IPCServer } from '../shared/ipc'
 import { logger } from '../shared/logger'
 import { JobWithToken, processJob } from './process'
+import { isJobSupported } from './shared'
 
 type PeerTubeServer = PeerTubeServerCommand & {
   runnerToken: string
@@ -169,18 +170,29 @@ export class RunnerServer {
   private async checkAvailableJobs () {
     if (this.checkingAvailableJobs) return
 
-    logger.info('Checking available jobs')
-
     this.checkingAvailableJobs = true
 
+    let hadAvailableJob = false
+
     for (const server of this.servers) {
       try {
+        logger.info('Checking available jobs on ' + server.url)
+
         const job = await this.requestJob(server)
         if (!job) continue
 
+        hadAvailableJob = true
+
         await this.tryToExecuteJobAsync(server, job)
       } catch (err) {
-        if ((err.res?.body as PeerTubeProblemDocument)?.code === ServerErrorCode.UNKNOWN_RUNNER_TOKEN) {
+        const code = (err.res?.body as PeerTubeProblemDocument)?.code
+
+        if (code === ServerErrorCode.RUNNER_JOB_NOT_IN_PROCESSING_STATE) {
+          logger.debug({ err }, 'Runner job is not in processing state anymore, retry later')
+          return
+        }
+
+        if (code === ServerErrorCode.UNKNOWN_RUNNER_TOKEN) {
           logger.error({ err }, `Unregistering ${server.url} as the runner token ${server.runnerToken} is invalid`)
 
           await this.unregisterRunner({ url: server.url })
@@ -192,6 +204,11 @@ export class RunnerServer {
     }
 
     this.checkingAvailableJobs = false
+
+    if (hadAvailableJob && this.canProcessMoreJobs()) {
+      this.checkAvailableJobs()
+        .catch(err => logger.error({ err }, 'Cannot check more available jobs'))
+    }
   }
 
   private async requestJob (server: PeerTubeServer) {
@@ -199,16 +216,18 @@ export class RunnerServer {
 
     const { availableJobs } = await server.runnerJobs.request({ runnerToken: server.runnerToken })
 
-    if (availableJobs.length === 0) {
+    const filtered = availableJobs.filter(j => isJobSupported(j))
+
+    if (filtered.length === 0) {
       logger.debug(`No job available on ${server.url} for runner ${server.runnerName}`)
       return undefined
     }
 
-    return availableJobs[0]
+    return filtered[0]
   }
 
   private async tryToExecuteJobAsync (server: PeerTubeServer, jobToAccept: { uuid: string }) {
-    if (this.processingJobs.length >= ConfigManager.Instance.getConfig().jobs.concurrency) return
+    if (!this.canProcessMoreJobs()) return
 
     const { job } = await server.runnerJobs.accept({ runnerToken: server.runnerToken, jobUUID: jobToAccept.uuid })
 
@@ -239,6 +258,10 @@ export class RunnerServer {
     return ConfigManager.Instance.setRegisteredInstances(data)
   }
 
+  private canProcessMoreJobs () {
+    return this.processingJobs.length < ConfigManager.Instance.getConfig().jobs.concurrency
+  }
+
   // ---------------------------------------------------------------------------
 
   private async cleanupTMP () {