More robust chunk handler

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

@@ -21,6 +21,9 @@ export class ProcessLiveRTMPHLSTranscoding {
   private readonly outputPath: string
   private readonly fsWatchers: FSWatcher[] = []
 
+  // Playlist name -> chunks
+  private readonly pendingChunksPerPlaylist = new Map<string, string[]>()
+
   private readonly playlistsCreated = new Set<string>()
   private allPlaylistsCreated = false
 
@@ -68,9 +71,19 @@ export class ProcessLiveRTMPHLSTranscoding {
           }
         })
 
-        tsWatcher.on('add', p => {
-          this.sendAddedChunkUpdate(p)
-            .catch(err => this.onUpdateError(err, rej))
+        tsWatcher.on('add', async p => {
+          try {
+            await this.sendPendingChunks()
+          } catch (err) {
+            this.onUpdateError(err, rej)
+          }
+
+          const playlistName = this.getPlaylistIdFromTS(p)
+
+          const pendingChunks = this.pendingChunksPerPlaylist.get(playlistName) || []
+          pendingChunks.push(p)
+
+          this.pendingChunksPerPlaylist.set(playlistName, pendingChunks)
         })
 
         tsWatcher.on('unlink', p => {
@@ -230,17 +243,19 @@ export class ProcessLiveRTMPHLSTranscoding {
     return this.updateWithRetry(payload)
   }
 
-  private sendAddedChunkUpdate (addedChunk: string): Promise<any> {
+  private async sendPendingChunks (): Promise<any> {
     if (this.ended) return Promise.resolve()
 
-    logger.debug(`Sending added live chunk ${addedChunk} update`)
+    for (const playlist of this.pendingChunksPerPlaylist.keys()) {
+      for (const chunk of this.pendingChunksPerPlaylist.get(playlist)) {
+        logger.debug(`Sending added live chunk ${chunk} update`)
 
-    const videoChunkFilename = basename(addedChunk)
+        const videoChunkFilename = basename(chunk)
 
         let payload: LiveRTMPHLSTranscodingUpdatePayload = {
           type: 'add-chunk',
           videoChunkFilename,
-      videoChunkFile: addedChunk
+          videoChunkFile: chunk
         }
 
         if (this.allPlaylistsCreated) {
@@ -254,7 +269,12 @@ export class ProcessLiveRTMPHLSTranscoding {
           }
         }
 
-    return this.updateWithRetry(payload)
+        this.updateWithRetry(payload)
+          .catch(err => logger.error({ err }, 'Cannot update with retry'))
+      }
+
+      this.pendingChunksPerPlaylist.set(playlist, [])
+    }
   }
 
   private async updateWithRetry (payload: LiveRTMPHLSTranscodingUpdatePayload, currentTry = 1): Promise<any> {
@@ -281,6 +301,12 @@ export class ProcessLiveRTMPHLSTranscoding {
     return `${videoChunkFilename.split('-')[0]}.m3u8`
   }
 
+  private getPlaylistIdFromTS (segmentPath: string) {
+    const playlistIdMatcher = /^([\d+])-/
+
+    return basename(segmentPath).match(playlistIdMatcher)[1]
+  }
+
   // ---------------------------------------------------------------------------
 
   private cleanup () {

server/lib/runners/job-handlers/live-rtmp-hls-transcoding-job-handler.ts

@@ -103,7 +103,7 @@ export class LiveRTMPHLSTranscodingJobHandler extends AbstractJobHandler<CreateO
       await move(updatePayload.masterPlaylistFile as string, join(outputDirectory, privatePayload.masterPlaylistName), { overwrite: true })
     }
 
-    logger.info(
+    logger.debug(
       'Runner live RTMP to HLS job %s for %s updated.',
       runnerJob.uuid, videoUUID, { updatePayload, ...this.lTags(videoUUID, runnerJob.uuid) }
     )

support/doc/tools.md

@@ -373,6 +373,7 @@ peertube-runner [commands] --id instance-2
 peertube-runner [commands] --id instance-3
 ```
 
+You can change the runner configuration (ffmpeg threads, ffmpeg nice etc) by editing `~/.config/peertube-runner-nodejs/[id]/config.toml`.
 
 ### Run the server