Enable external plugins to test the PR

.github/workflows/test.yml

@@ -77,7 +77,7 @@ jobs:
 
       - name: Run Test
         # external-plugins tests only run on schedule
-        if: github.event_name == 'schedule' || matrix.test_suite != 'external-plugins'
+        # if: github.event_name == 'schedule' || matrix.test_suite != 'external-plugins'
         env:
           AKISMET_KEY: ${{ secrets.AKISMET_KEY }}
         run: npm run ci -- ${{ matrix.test_suite }}

client/src/app/+admin/system/runners/runner-job-list/runner-job-list.component.html

@@ -84,7 +84,7 @@
           <pre>{{ runnerJob.privatePayload }}</pre>
         </div>
 
-        <pre *ngIf="runnerJob.error" class=".text-danger" >{{ runnerJob.error }}</pre>
+        <pre *ngIf="runnerJob.error" class="text-danger" >{{ runnerJob.error }}</pre>
       </td>
     </tr>
   </ng-template>

packages/peertube-runner/README.md

@@ -1 +1,23 @@
 # PeerTube runner
+
+Runner program to execute jobs (transcoding...) of remote PeerTube instances.
+
+Commands below has to be run at the root of PeerTube git repository.
+
+## Develop
+
+```bash
+npm run dev:peertube-runner
+```
+
+## Build
+
+```bash
+npm run build:peertube-runner
+```
+
+## Run
+
+```bash
+node packages/peertube-runner/dist/peertube-runner.js --help
+```

packages/peertube-runner/peertube-runner.ts

@@ -26,7 +26,7 @@ program.command('server')
     try {
       await RunnerServer.Instance.run()
     } catch (err) {
-      console.error('Cannot run PeerTube runner as server mode', err)
+      logger.error('Cannot run PeerTube runner as server mode', err)
       process.exit(-1)
     }
   })
@@ -41,7 +41,7 @@ program.command('register')
     try {
       await registerRunner(options)
     } catch (err) {
-      console.error('Cannot register this PeerTube runner.', err)
+      logger.error('Cannot register this PeerTube runner.', err)
       process.exit(-1)
     }
   })
@@ -53,7 +53,7 @@ program.command('unregister')
     try {
       await unregisterRunner(options)
     } catch (err) {
-      console.error('Cannot unregister this PeerTube runner.', err)
+      logger.error('Cannot unregister this PeerTube runner.', err)
       process.exit(-1)
     }
   })
@@ -64,7 +64,7 @@ program.command('list-registered')
     try {
       await listRegistered()
     } catch (err) {
-      console.error('Cannot list registered PeerTube instances.', err)
+      logger.error('Cannot list registered PeerTube instances.', err)
       process.exit(-1)
     }
   })

packages/peertube-runner/server/process/shared/process-live.ts

@@ -204,8 +204,8 @@ export class ProcessLiveRTMPHLSTranscoding {
 
   // ---------------------------------------------------------------------------
 
-  private sendDeletedChunkUpdate (deletedChunk: string) {
+  private sendDeletedChunkUpdate (deletedChunk: string): Promise<any> {
-    if (this.ended) return
+    if (this.ended) return Promise.resolve()
 
     logger.debug(`Sending removed live chunk ${deletedChunk} update`)
 
@@ -230,8 +230,8 @@ export class ProcessLiveRTMPHLSTranscoding {
     return this.updateWithRetry(payload)
   }
 
-  private sendAddedChunkUpdate (addedChunk: string) {
+  private sendAddedChunkUpdate (addedChunk: string): Promise<any> {
-    if (this.ended) return
+    if (this.ended) return Promise.resolve()
 
     logger.debug(`Sending added live chunk ${addedChunk} update`)
 
@@ -257,7 +257,7 @@ export class ProcessLiveRTMPHLSTranscoding {
     return this.updateWithRetry(payload)
   }
 
-  private async updateWithRetry (payload: LiveRTMPHLSTranscodingUpdatePayload, currentTry = 1) {
+  private async updateWithRetry (payload: LiveRTMPHLSTranscodingUpdatePayload, currentTry = 1): Promise<any> {
     if (this.ended || this.errored) return
 
     try {

packages/peertube-runner/server/server.ts

@@ -23,6 +23,8 @@ export class RunnerServer {
 
   private checkingAvailableJobs = false
 
+  private cleaningUp = false
+
   private readonly sockets = new Map<PeerTubeServer, Socket>()
 
   private constructor () {}
@@ -45,13 +47,17 @@ export class RunnerServer {
     try {
       await ipcServer.run(this)
     } catch (err) {
-      console.error('Cannot start local socket for IPC communication', err)
+      logger.error('Cannot start local socket for IPC communication', err)
       process.exit(-1)
     }
 
     // Cleanup on exit
     for (const code of [ 'SIGINT', 'SIGUSR1', 'SIGUSR2', 'uncaughtException' ]) {
-      process.on(code, async () => {
+      process.on(code, async (err, origin) => {
+        if (code === 'uncaughtException') {
+          logger.error({ err, origin }, 'uncaughtException')
+        }
+
         await this.onExit()
       })
     }
@@ -244,6 +250,11 @@ export class RunnerServer {
   }
 
   private async onExit () {
+    if (this.cleaningUp) return
+    this.cleaningUp = true
+
+    logger.info('Cleaning up after program exit')
+
     try {
       for (const { server, job } of this.processingJobs) {
         await server.runnerJobs.abort({
@@ -256,7 +267,7 @@ export class RunnerServer {
 
       await this.cleanupTMP()
     } catch (err) {
-      console.error(err)
+      logger.error(err)
       process.exit(-1)
     }
 

packages/peertube-runner/shared/http.ts

@@ -47,7 +47,7 @@ export function downloadFile (options: {
 
     request.on('error', err => {
       remove(destination)
-        .catch(err => console.error(err))
+        .catch(err => logger.error(err))
 
       return rej(err)
     })

packages/peertube-runner/shared/ipc/ipc-server.ts

@@ -27,7 +27,7 @@ export class IPCServer {
 
         this.sendReponse(res, { success: true, data })
       } catch (err) {
-        console.error('Cannot execute RPC call', err)
+        logger.error('Cannot execute RPC call', err)
         this.sendReponse(res, { success: false, error: err.message })
       }
     })
@@ -56,6 +56,6 @@ export class IPCServer {
     body: IPCReponse<T>
   ) {
     response(body)
-      .catch(err => console.error('Cannot send response after IPC request', err))
+      .catch(err => logger.error('Cannot send response after IPC request', err))
   }
 }

scripts/build/peertube-runner.sh

@@ -10,4 +10,4 @@ rm -rf ./dist
 rm -rf ./dist
 mkdir ./dist
 
-./node_modules/.bin/esbuild ./peertube-runner.ts --bundle --platform=node --external:"./lib-cov/fluent-ffmpeg" --external:pg-hstore --outfile=dist/peertube-runner.js
+./node_modules/.bin/esbuild ./peertube-runner.ts --bundle --platform=node --target=node14 --external:"./lib-cov/fluent-ffmpeg" --external:pg-hstore --outfile=dist/peertube-runner.js

server/lib/job-queue/job-queue.ts

@@ -96,6 +96,7 @@ export type CreateJobArgument =
 export type CreateJobOptions = {
   delay?: number
   priority?: number
+  failParentOnFailure?: boolean
 }
 
 const handlers: { [id in JobType]: (job: Job) => Promise<any> } = {
@@ -363,7 +364,11 @@ class JobQueue {
       name: 'job',
       data: job.payload,
       queueName: job.type,
-      opts: this.buildJobOptions(job.type as JobType, pick(job, [ 'priority', 'delay' ]))
+      opts: {
+        failParentOnFailure: true,
+
+        ...this.buildJobOptions(job.type as JobType, pick(job, [ 'priority', 'delay', 'failParentOnFailure' ]))
+      }
     }
   }
 

server/lib/live/shared/muxing-session.ts

@@ -79,9 +79,7 @@ class MuxingSession extends EventEmitter {
   private streamingPlaylist: MStreamingPlaylistVideo
   private liveSegmentShaStore: LiveSegmentShaStore
 
-  private tsWatcher: FSWatcher
+  private filesWatcher: FSWatcher
-  private masterWatcher: FSWatcher
-  private m3u8Watcher: FSWatcher
 
   private masterPlaylistCreated = false
   private liveReady = false
@@ -149,6 +147,8 @@ class MuxingSession extends EventEmitter {
 
     await this.transcodingWrapper.run()
 
+    this.filesWatcher = watch(this.outDirectory, { depth: 0 })
+
     this.watchMasterFile()
     this.watchTSFiles()
     this.watchM3U8File()
@@ -168,9 +168,10 @@ class MuxingSession extends EventEmitter {
   }
 
   private watchMasterFile () {
-    this.masterWatcher = watch(this.outDirectory + '/' + this.streamingPlaylist.playlistFilename)
+    this.filesWatcher.on('add', async path => {
+      if (path !== join(this.outDirectory, this.streamingPlaylist.playlistFilename)) return
+      if (this.masterPlaylistCreated === true) return
 
-    this.masterWatcher.on('add', async () => {
       try {
         if (this.streamingPlaylist.storage === VideoStorage.OBJECT_STORAGE) {
           const url = await storeHLSFileFromFilename(this.streamingPlaylist, this.streamingPlaylist.playlistFilename)
@@ -188,20 +189,18 @@ class MuxingSession extends EventEmitter {
       this.masterPlaylistCreated = true
 
       logger.info('Master playlist file for %s has been created', this.videoUUID, this.lTags())
-
-      this.masterWatcher.close()
-        .catch(err => logger.error('Cannot close master watcher of %s.', this.outDirectory, { err, ...this.lTags() }))
     })
   }
 
   private watchM3U8File () {
-    this.m3u8Watcher = watch(this.outDirectory + '/*.m3u8')
-
     const sendQueues = new Map<string, PQueue>()
 
-    const onChangeOrAdd = async (m3u8Path: string) => {
+    const onChange = async (m3u8Path: string) => {
+      if (m3u8Path.endsWith('.m3u8') !== true) return
       if (this.streamingPlaylist.storage !== VideoStorage.OBJECT_STORAGE) return
 
+      logger.debug('Live change handler of M3U8 file %s.', m3u8Path, this.lTags())
+
       try {
         if (!sendQueues.has(m3u8Path)) {
           sendQueues.set(m3u8Path, new PQueue({ concurrency: 1 }))
@@ -214,18 +213,18 @@ class MuxingSession extends EventEmitter {
       }
     }
 
-    this.m3u8Watcher.on('change', onChangeOrAdd)
+    this.filesWatcher.on('change', onChange)
   }
 
   private watchTSFiles () {
     const startStreamDateTime = new Date().getTime()
 
-    this.tsWatcher = watch(this.outDirectory + '/*.ts')
-
     const playlistIdMatcher = /^([\d+])-/
 
     const addHandler = async (segmentPath: string) => {
-      logger.debug('Live add handler of %s.', segmentPath, this.lTags())
+      if (segmentPath.endsWith('.ts') !== true) return
+
+      logger.debug('Live add handler of TS file %s.', segmentPath, this.lTags())
 
       const playlistId = basename(segmentPath).match(playlistIdMatcher)[0]
 
@@ -252,6 +251,10 @@ class MuxingSession extends EventEmitter {
     }
 
     const deleteHandler = async (segmentPath: string) => {
+      if (segmentPath.endsWith('.ts') !== true) return
+
+      logger.debug('Live delete handler of TS file %s.', segmentPath, this.lTags())
+
       try {
         await this.liveSegmentShaStore.removeSegmentSha(segmentPath)
       } catch (err) {
@@ -267,8 +270,8 @@ class MuxingSession extends EventEmitter {
       }
     }
 
-    this.tsWatcher.on('add', p => addHandler(p))
+    this.filesWatcher.on('add', p => addHandler(p))
-    this.tsWatcher.on('unlink', p => deleteHandler(p))
+    this.filesWatcher.on('unlink', p => deleteHandler(p))
   }
 
   private async isQuotaExceeded (segmentPath: string) {
@@ -371,7 +374,8 @@ class MuxingSession extends EventEmitter {
     setTimeout(() => {
       // Wait latest segments generation, and close watchers
 
-      Promise.all([ this.tsWatcher.close(), this.masterWatcher.close(), this.m3u8Watcher.close() ])
+      const promise = this.filesWatcher?.close() || Promise.resolve()
+      promise
         .then(() => {
           // Process remaining segments hash
           for (const key of Object.keys(this.segmentsToProcessPerPlaylist)) {

server/lib/runners/job-handlers/abstract-job-handler.ts

@@ -21,6 +21,7 @@ import {
   RunnerJobVODWebVideoTranscodingPayload,
   RunnerJobVODWebVideoTranscodingPrivatePayload
 } from '@shared/models'
+import { throttle } from 'lodash'
 
 type CreateRunnerJobArg =
   {
@@ -48,6 +49,8 @@ export abstract class AbstractJobHandler <C, U extends RunnerJobUpdatePayload, S
 
   protected readonly lTags = loggerTagsFactory('runner')
 
+  static setJobAsUpdatedThrottled = throttle(setAsUpdated, 2000)
+
   // ---------------------------------------------------------------------------
 
   abstract create (options: C): Promise<MRunnerJob>
@@ -102,16 +105,19 @@ export abstract class AbstractJobHandler <C, U extends RunnerJobUpdatePayload, S
 
     if (progress) runnerJob.progress = progress
 
-    await retryTransactionWrapper(() => {
+    if (!runnerJob.changed()) {
-      return sequelizeTypescript.transaction(async transaction => {
+      try {
-        if (runnerJob.changed()) {
+        await AbstractJobHandler.setJobAsUpdatedThrottled({ sequelize: sequelizeTypescript, table: 'runnerJob', id: runnerJob.id })
-          return runnerJob.save({ transaction })
+      } catch (err) {
+        logger.warn('Cannot set remote job as updated', { err, ...this.lTags(runnerJob.id, runnerJob.type) })
       }
 
-        // Don't update the job too often
+      return
-        if (new Date().getTime() - runnerJob.updatedAt.getTime() > 2000) {
-          await setAsUpdated({ sequelize: sequelizeTypescript, table: 'runnerJob', id: runnerJob.id, transaction })
     }
+
+    await retryTransactionWrapper(() => {
+      return sequelizeTypescript.transaction(async transaction => {
+        return runnerJob.save({ transaction })
       })
     })
   }

server/tests/api/videos/resumable-upload.ts

@@ -79,7 +79,7 @@ describe('Test resumable upload', function () {
   async function checkFileSize (uploadIdArg: string, expectedSize: number | null) {
     const uploadId = uploadIdArg.replace(/^upload_id=/, '')
 
-    const subPath = join('tmp', 'resumable-uploads', uploadId)
+    const subPath = join('tmp', 'resumable-uploads', `${rootId}-${uploadId}.mp4`)
     const filePath = server.servers.buildDirectory(subPath)
     const exists = await pathExists(filePath)
 

server/tests/cli/create-import-video-file-job.ts

@@ -73,7 +73,7 @@ function runTests (objectStorage: boolean) {
   })
 
   it('Should run a import job on video 1 with a lower resolution', async function () {
-    const command = `npm run create-import-video-file-job -- -v ${video1ShortId} -i server/tests/fixtures/video_short-480.webm`
+    const command = `npm run create-import-video-file-job -- -v ${video1ShortId} -i server/tests/fixtures/video_short_480.webm`
     await servers[0].cli.execWithEnv(command)
 
     await waitJobs(servers)

shared/server-commands/requests/requests.ts

@@ -159,7 +159,7 @@ function unwrapBodyOrDecodeToJSON <T> (test: request.Test): Promise<T> {
       try {
         return JSON.parse(new TextDecoder().decode(res.body))
       } catch (err) {
-        console.error('Cannot decode JSON.', res.body instanceof Buffer ? res.body.toString() : res.body)
+        console.error('Cannot decode JSON.', { res, body: res.body instanceof Buffer ? res.body.toString() : res.body })
         throw err
       }
     }
@@ -168,7 +168,7 @@ function unwrapBodyOrDecodeToJSON <T> (test: request.Test): Promise<T> {
       try {
         return JSON.parse(res.text)
       } catch (err) {
-        console.error('Cannot decode json', res.text)
+        console.error('Cannot decode json', { res, text: res.text })
         throw err
       }
     }