From d5cf99114ac5ad3754e14e4aa7b5e0f2f1303dbd Mon Sep 17 00:00:00 2001 From: Eli Mallon Date: Tue, 15 Sep 2026 20:27:56 -0700 Subject: [PATCH] transcoding: fresh gateway manifest after five refused pushes transcoding: a stream moves to a fresh gateway manifest after five refused pushes in a row. Production stopped transcoding with every push answered 503 "No sessions available" on both live streams, while the same gateway, at the same time, transcoded our TS of those very streams under a fresh manifest with the same profiles (all renditions, all 200). 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 over stays fallen over for the life of the stream, and pushing to it forever keeps it alive. Now RotateAfterFailures consecutive refusals (a non-200 or a failed request) pick a new manifest trailer and restart the sequence at 0; a success resets the run; streamplace_transcode_manifest_rotations_total counts it. The manifest and sequence are now claimed under a lock when the push slot is taken (the sequence used to race between the two in-flight pushes, which showed as out-of-order pushes in the gateway log), and a malformed multipart response no longer dereferences a nil part before its error is checked Claude-Session: https://claude.ai/code/session_014aPQ5yqG9QFxKnbwQfCYKa (cherry picked from commit 633c0531f1470ff4f552987dece1b54a215b1ece) --- pkg/livepeer/livepeer.go | 88 ++++++++++++++++++++++++++++------- pkg/livepeer/rotate_test.go | 92 +++++++++++++++++++++++++++++++++++++ pkg/spmetrics/spmetrics.go | 5 ++ 3 files changed, 168 insertions(+), 17 deletions(-) create mode 100644 pkg/livepeer/rotate_test.go 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", -- knotmirror 0.0.1