From def4ea4f3824689abd4ea66f5ab6e47d553b7ddd Mon Sep 17 00:00:00 2001
From: Chocobozzz <me@florianbigard.com>
Date: Fri, 19 May 2023 13:46:40 +0200
Subject: More robust chunk handler

---
 .../server/process/shared/process-live.ts          | 66 +++++++++++++++-------
 .../live-rtmp-hls-transcoding-job-handler.ts       |  2 +-
 support/doc/tools.md                               |  1 +
 3 files changed, 48 insertions(+), 21 deletions(-)

diff --git a/packages/peertube-runner/server/process/shared/process-live.ts b/packages/peertube-runner/server/process/shared/process-live.ts
index b17b51c7c..df1b677f0 100644
--- a/packages/peertube-runner/server/process/shared/process-live.ts
+++ b/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,31 +243,38 @@ 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
-    }
+        let payload: LiveRTMPHLSTranscodingUpdatePayload = {
+          type: 'add-chunk',
+          videoChunkFilename,
+          videoChunkFile: chunk
+        }
 
-    if (this.allPlaylistsCreated) {
-      const playlistName = this.getPlaylistName(videoChunkFilename)
+        if (this.allPlaylistsCreated) {
+          const playlistName = this.getPlaylistName(videoChunkFilename)
 
-      payload = {
-        ...payload,
-        masterPlaylistFile: join(this.outputPath, 'master.m3u8'),
-        resolutionPlaylistFilename: playlistName,
-        resolutionPlaylistFile: join(this.outputPath, playlistName)
+          payload = {
+            ...payload,
+            masterPlaylistFile: join(this.outputPath, 'master.m3u8'),
+            resolutionPlaylistFilename: playlistName,
+            resolutionPlaylistFile: join(this.outputPath, playlistName)
+          }
+        }
+
+        this.updateWithRetry(payload)
+          .catch(err => logger.error({ err }, 'Cannot update with retry'))
       }
-    }
 
-    return this.updateWithRetry(payload)
+      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 () {
diff --git a/server/lib/runners/job-handlers/live-rtmp-hls-transcoding-job-handler.ts b/server/lib/runners/job-handlers/live-rtmp-hls-transcoding-job-handler.ts
index c3b0a95cc..87b6f0702 100644
--- a/server/lib/runners/job-handlers/live-rtmp-hls-transcoding-job-handler.ts
+++ b/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) }
     )
diff --git a/support/doc/tools.md b/support/doc/tools.md
index 5ddef5c5a..c12b00e32 100644
--- a/support/doc/tools.md
+++ b/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
 
-- 
cgit v1.2.3