Fix RTMP pull (#3377)

* Fix RTMP pull

* Retries for StreamExists check

* add log line

media/rtmp2segment.go

@@ -18,6 +18,7 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/cenkalti/backoff"
 	"github.com/livepeer/go-livepeer/clog"
 	"golang.org/x/sys/unix"
 )
@@ -46,16 +47,23 @@ func (ms *MediaSegmenter) RunSegmentation(ctx context.Context, in string, segmen
 		processSegments(ctx, segmentHandler, outFilePattern, completionSignal)
 	}()
 
-	retryCount := 0
 	for {
-		streamExists, err := ms.MediaMTXClient.StreamExists()
+		err := backoff.Retry(func() error {
+			streamExists, err := ms.MediaMTXClient.StreamExists()
+			if err != nil {
+				return fmt.Errorf("StreamExists check failed: %w", err)
+			}
+			if !streamExists {
+				clog.Errorf(ctx, "input stream does not exist")
+				return fmt.Errorf("input stream does not exist")
+			}
+			return nil
+		}, backoff.WithMaxRetries(newExponentialBackOff(), 3))
 		if err != nil {
-			clog.Errorf(ctx, "StreamExists check failed. err=%s", err)
-		}
-		if retryCount > 2 && !streamExists {
-			clog.Errorf(ctx, "Stopping segmentation, input stream does not exist. in=%s err=%s", in, err)
+			clog.Errorf(ctx, "Stopping segmentation in=%s err=%s", in, err)
 			break
 		}
+		clog.Infof(ctx, "Starting segmentation. in=%s", in)
 		cmd := exec.CommandContext(procCtx, "ffmpeg",
 			"-i", in,
 			"-c:a", "copy",
@@ -66,17 +74,24 @@ func (ms *MediaSegmenter) RunSegmentation(ctx context.Context, in string, segmen
 		output, err := cmd.CombinedOutput()
 		if err != nil {
 			clog.Errorf(ctx, "Error receiving RTMP: %v", err)
-			clog.Infof(ctx, "Process output: %s", output)
-			return
+			break
 		}
-		retryCount++
+		clog.Infof(ctx, "Segmentation ffmpeg output: %s", output)
 		time.Sleep(5 * time.Second)
 	}
 	completionSignal <- true
 	clog.Infof(ctx, "sent completion signal, now waiting")
 	wg.Wait()
 }
 
+func newExponentialBackOff() *backoff.ExponentialBackOff {
+	backOff := backoff.NewExponentialBackOff()
+	backOff.InitialInterval = 500 * time.Millisecond
+	backOff.MaxInterval = 5 * time.Second
+	backOff.Reset()
+	return backOff
+}
+
 func createNamedPipe(pipeName string) {
 	err := syscall.Mkfifo(pipeName, 0666)
 	if err != nil && !os.IsExist(err) {

server/ai_live_video.go

@@ -205,6 +205,7 @@ func startTrickleSubscribe(ctx context.Context, url *url.URL, params aiRequestPa
 			}
 		}()
 		for {
+			clog.V(6).Infof(ctx, "Starting output rtmp")
 			if !params.inputStreamExists() {
 				clog.Errorf(ctx, "Stopping output rtmp stream, input stream does not exist.")
 				break