diff --git a/pkg/livepeer/livepeer.go b/pkg/livepeer/livepeer.go index ff888e702..ab58ef89c 100644 --- a/pkg/livepeer/livepeer.go +++ b/pkg/livepeer/livepeer.go @@ -13,6 +13,7 @@ import ( "net/http" "os" "strings" + "sync" "sync/atomic" "time" @@ -38,13 +39,28 @@ const MaxWaiting = 2 // waiting for a transcode slot. var ErrBacklog = errors.New("transcode backlog: segment skipped") +// RotateAfterFailures is how many pushes in a row the gateway may refuse +// before the session moves to a fresh manifest ID. A go-livepeer gateway +// keeps per-manifest state (its orchestrator sessions and suspensions) for +// as long as segments keep arriving under that manifest, so a manifest that +// has fallen into "No sessions available" stays there forever while a fresh +// one on the same gateway transcodes fine: that is exactly what a stuck +// production stream looked like. A new manifest is a new stream to the +// gateway; the segment sequence restarts at 0 with it. +const RotateAfterFailures = 5 + type LivepeerSession struct { + // SessionID is the manifest ID's base: "-", a new + // trailer per rotation. Guarded by mu, like Count. SessionID string Count int GatewayURL string Guard chan struct{} CLI *config.CLI waiting atomic.Int32 + mu sync.Mutex + did string + failures int } // borrowed from catalyst-api @@ -59,36 +75,61 @@ func RandomTrailer(length int) string { } func NewLivepeerSession(ctx context.Context, cli *config.CLI, did string, gatewayURL string) (*LivepeerSession, error) { - sessionID := fmt.Sprintf("%s-%s", did, RandomTrailer(8)) - sessionID = strings.ReplaceAll(sessionID, ":", "") - sessionID = strings.ReplaceAll(sessionID, ".", "") return &LivepeerSession{ - SessionID: sessionID, + SessionID: manifestBase(did), Count: 0, GatewayURL: gatewayURL, Guard: make(chan struct{}, SegmentsInFlight), CLI: cli, + did: did, }, nil } +func manifestBase(did string) string { + sessionID := fmt.Sprintf("%s-%s", did, RandomTrailer(8)) + sessionID = strings.ReplaceAll(sessionID, ":", "") + return strings.ReplaceAll(sessionID, ".", "") +} + +// next claims the manifest ID and sequence number for one push. +func (ls *LivepeerSession) next(numRenditions int) (string, int) { + ls.mu.Lock() + defer ls.mu.Unlock() + seq := ls.Count + ls.Count++ + return fmt.Sprintf("%s-%dren", ls.SessionID, numRenditions), seq +} + +// noteResult records whether the gateway took a push; RotateAfterFailures +// refusals in a row move the session to a fresh manifest. +func (ls *LivepeerSession) noteResult(ctx context.Context, ok bool) { + ls.mu.Lock() + defer ls.mu.Unlock() + if ok { + ls.failures = 0 + return + } + ls.failures++ + if ls.failures < RotateAfterFailures { + return + } + old := ls.SessionID + ls.SessionID = manifestBase(ls.did) + ls.Count = 0 + ls.failures = 0 + spmetrics.TranscodeManifestRotationsTotal.Inc() + log.Warn(ctx, "transcode: gateway refused every recent push, moving to a fresh manifest", "failures", RotateAfterFailures, "old", old, "new", ls.SessionID) +} + func (ls *LivepeerSession) PostSegmentToGateway(ctx context.Context, buf []byte, spseg *placestream.Segment, rs renditions.Renditions) ([][]byte, error) { ctx = log.WithLogValues(ctx, "func", "PostSegmentToGateway") lpProfiles := rs.ToLivepeerProfiles() - sessionIDRen := fmt.Sprintf("%s-%dren", ls.SessionID, len(rs)) - transcodingConfiguration := map[string]any{ - "manifestID": sessionIDRen, - "profiles": lpProfiles, - } - bs, err := json.Marshal(transcodingConfiguration) - if err != nil { - return nil, fmt.Errorf("failed to marshal livepeer profile: %w", err) - } tsSeg := bytes.Buffer{} audioSeg := bytes.Buffer{} // Bounded: a conversion pipeline that never reaches EOS must not hold // the in-flight guard forever and silently stop every later transcode. convCtx, convCancel := context.WithTimeout(ctx, 30*time.Second) - err = media.MP4ToMPEGTSVideoMP4Audio(convCtx, bytes.NewReader(buf), &tsSeg, &audioSeg) + err := media.MP4ToMPEGTSVideoMP4Audio(convCtx, bytes.NewReader(buf), &tsSeg, &audioSeg) convCancel() if err != nil { return nil, fmt.Errorf("failed to convert mp4 to ts video/mp4 audio: %w", err) @@ -113,9 +154,19 @@ func (ls *LivepeerSession) PostSegmentToGateway(ctx context.Context, buf []byte, } ctx, cancel := context.WithTimeout(ctx, time.Minute*5) defer cancel() - seqNo := ls.Count + // Claimed only now, with the slot: the manifest may have rotated while + // this segment waited. + sessionIDRen, seqNo := ls.next(len(rs)) + transcodingConfiguration := map[string]any{ + "manifestID": sessionIDRen, + "profiles": lpProfiles, + } + bs, err := json.Marshal(transcodingConfiguration) + if err != nil { + <-ls.Guard + return nil, fmt.Errorf("failed to marshal livepeer profile: %w", err) + } url := fmt.Sprintf("%s/live/%s/%d.ts", ls.GatewayURL, sessionIDRen, seqNo) - ls.Count++ dur := time.Duration(*spseg.Duration) durationMs := int(dur.Milliseconds()) @@ -161,6 +212,7 @@ func (ls *LivepeerSession) PostSegmentToGateway(ctx context.Context, buf []byte, resp, err := aqhttp.DoTrusted(ctx, req) if err != nil { <-ls.Guard + ls.noteResult(ctx, false) return nil, fmt.Errorf("failed to send segment to gateway (config %s): %w", string(bs), err) } <-ls.Guard @@ -168,8 +220,10 @@ func (ls *LivepeerSession) PostSegmentToGateway(ctx context.Context, buf []byte, if resp.StatusCode != http.StatusOK { errOut, _ := io.ReadAll(resp.Body) + ls.noteResult(ctx, false) return nil, fmt.Errorf("gateway returned non-OK status (config %s): %d, %s", string(bs), resp.StatusCode, string(errOut)) } + ls.noteResult(ctx, true) var out [][]byte @@ -184,10 +238,10 @@ func (ls *LivepeerSession) PostSegmentToGateway(ctx context.Context, buf []byte, if err == io.EOF { break } - ctx := log.WithLogValues(ctx, "part", p.FileName()) if err != nil { return nil, fmt.Errorf("failed to get next part: %w", err) } + ctx := log.WithLogValues(ctx, "part", p.FileName()) mp4Bs := bytes.Buffer{} audioReader := bytes.NewReader(audioSeg.Bytes()) if ls.CLI.LivepeerDebug { diff --git a/pkg/livepeer/rotate_test.go b/pkg/livepeer/rotate_test.go new file mode 100644 index 000000000..25f22a31d --- /dev/null +++ b/pkg/livepeer/rotate_test.go @@ -0,0 +1,92 @@ +package livepeer + +import ( + "context" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "runtime" + "sync" + "testing" + + "github.com/stretchr/testify/require" + "stream.place/streamplace/pkg/config" + "stream.place/streamplace/pkg/placestream" + "stream.place/streamplace/pkg/renditions" +) + +// A gateway that has fallen into "No sessions available" for a manifest +// stays there while pushes keep coming under it; after RotateAfterFailures +// refusals in a row the session moves to a fresh manifest, whose sequence +// starts at 0 again, and that one is taken. +func TestRotatesManifestAfterRefusals(t *testing.T) { + _, file, _, _ := runtime.Caller(0) + seg, err := os.ReadFile(filepath.Join(filepath.Dir(file), "..", "..", "test", "fixtures", "h264-opus-frag.mp4")) + require.NoError(t, err) + + var mu sync.Mutex + var seen []string // manifest/seq per push + var stuck string // the first manifest: refused forever + gw := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + defer mu.Unlock() + // /live//.ts + rest := r.URL.Path[len("/live/"):] + seen = append(seen, rest) + manifest := rest[:len(rest)-len(filepath.Base(rest))-1] + if stuck == "" { + stuck = manifest + } + if manifest == stuck { + w.WriteHeader(http.StatusServiceUnavailable) + _, _ = w.Write([]byte("No sessions available\n")) + return + } + w.Header().Set("Content-Type", "multipart/mixed; boundary=x") + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte("--x--\r\n")) // no parts + })) + defer gw.Close() + + ctx := context.Background() + ls, err := NewLivepeerSession(ctx, &config.CLI{}, "did:plc:test", gw.URL) + require.NoError(t, err) + first := ls.SessionID + dur := int64(1e9) + spseg := &placestream.Segment{Creator: "did:plc:test", Duration: &dur, Video: []placestream.Segment_Video{{Width: 320, Height: 240}}} + rs := renditions.Renditions{{Name: "160p", Width: 284, Height: 160, Bitrate: 250000}} + + for i := 0; i < RotateAfterFailures; i++ { + _, err := ls.PostSegmentToGateway(ctx, seg, spseg, rs) + require.Error(t, err, "push %d is refused", i) + require.Contains(t, err.Error(), "No sessions available") + } + require.NotEqual(t, first, ls.SessionID, "moved to a fresh manifest") + require.Equal(t, 0, ls.Count, "the sequence restarts with it") + + _, err = ls.PostSegmentToGateway(ctx, seg, spseg, rs) + require.NoError(t, err, "the fresh manifest is taken") + mu.Lock() + defer mu.Unlock() + require.Len(t, seen, RotateAfterFailures+1) + require.Equal(t, first+"-1ren/0.ts", seen[0]) + require.Equal(t, first+"-1ren/4.ts", seen[RotateAfterFailures-1]) + require.Equal(t, ls.SessionID+"-1ren/0.ts", seen[RotateAfterFailures]) +} + +func TestNoteResultResetsOnSuccess(t *testing.T) { + ls, err := NewLivepeerSession(context.Background(), &config.CLI{}, "did:plc:test", "http://gw") + require.NoError(t, err) + first := ls.SessionID + for i := 0; i < RotateAfterFailures-1; i++ { + ls.noteResult(context.Background(), false) + } + ls.noteResult(context.Background(), true) + for i := 0; i < RotateAfterFailures-1; i++ { + ls.noteResult(context.Background(), false) + } + require.Equal(t, first, ls.SessionID, "a success in between resets the run") + ls.noteResult(context.Background(), false) + require.NotEqual(t, first, ls.SessionID) +} diff --git a/pkg/spmetrics/spmetrics.go b/pkg/spmetrics/spmetrics.go index 44865a312..dbcf4d060 100644 --- a/pkg/spmetrics/spmetrics.go +++ b/pkg/spmetrics/spmetrics.go @@ -56,6 +56,11 @@ var TranscodeErrorsTotal = promauto.NewCounter(prometheus.CounterOpts{ Help: "total number of transcode errors", }) +var TranscodeManifestRotationsTotal = promauto.NewCounter(prometheus.CounterOpts{ + Name: "streamplace_transcode_manifest_rotations_total", + Help: "times a stream moved to a fresh gateway manifest after the gateway refused every recent push (see livepeer.RotateAfterFailures)", +}) + var TranscodeDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{ Name: "streamplace_transcode_duration_ms", Help: "duration of transcode in ms",