ai/live: Gateway-native WHEP server.

Adds a native WHEP server to the gateway for playback. This
offers:

  * Immediate connectivity. WHEP playback can begin as soon as
    the WHIP response is received, rather than having to poll
    the WHEP endpoint repeatedly until the output is available.
    The WHEP response will simply hang until the output is ready.

  * Faster startup times, especially in no-audio cases. We use the
    mpegts PMT to know immediately which tracks are available. This
    avoids having to probe several seconds for an audio track that
    may never come.

  * Removes the requirement to have a MediaMTX server in the mix.
    This can substantially simplify the infrastructure requirements
    for many deployments. However, MediaMTX would still be needed for
    RTMP ingest or non-WHEP playback.

  * The promise of much finer control over tracks and stats reports.

/// Detailed Changes

  * Adds a WHEP server. Supports video (H.264) and audio (Opus).

  * Adds a mpegts-to-RTP timestamp converter to handle roll-overs
    between 33-bit mpegts (ugh) and 32-bit RTP timestamps.

  * Adds the output RingBuffer to the LivePipelines struct so it can be
    accessible via the WHEP handler. If the output is not yet ready,
    the WHEP handler waits until it is (via condvar). This allows WHEP
    sessions to be set up immediately as soon as ingest begins,
    regardless of whether the output is ready yet.

  * Updates the output RingBuffer to follow the io.Closer interface.
    Needed so we close the output buffer instead of writing to it
    indefinitely after a WHEP peerconnection goes away.

  * Off by default. The `LIVE_AI_WHEP_ADDR` environment variable
    needs to be set with at least a port number. This should be
    different from the WHIP port at `LIVE_AI_WHIP_ADDR`.

  * MediaMTX playback is still available; nothing changes there.

  * The swapping behavior does not change for now. If we swap
    orchestrators, the WHEP peerconnection will close and the client
    will have to re-connect WHEP (swaps do not affect ingest).

/// Rollout Plan

  * Enable on staging via configuring a port via the
    `LIVE_AI_WHEP_ADDR` env var. Update `LIVE_AI_WHEP_URL` to
    point to the gateway instead of MediaMTX.

  * Leave things on staging for a couple weeks

  * When we are satisfied with staging, enable on production.

Note that we can always enable WHEP on the gateway for manual testing
without making it the playback default as long as the `LIVE_AI_WHEP_URL`
env var points to MediaMTX.

Author: j0sh

core/livepeernode.go

@@ -19,6 +19,7 @@ import (
 	"time"
 
 	"github.com/golang/glog"
+	"github.com/livepeer/go-livepeer/media"
 	"github.com/livepeer/go-livepeer/pm"
 	"github.com/livepeer/go-livepeer/trickle"
 
@@ -183,6 +184,9 @@ type LivePipeline struct {
 	ControlPub   *trickle.TricklePublisher
 	StopControl  func()
 	ReportUpdate func([]byte)
+	OutCond      *sync.Cond
+	OutWriter    *media.RingBuffer
+	Closed       bool
 }
 
 // NewLivepeerNode creates a new Livepeer Node. Eth can be nil.

go.mod

@@ -1,6 +1,6 @@
 module github.com/livepeer/go-livepeer
 
-go 1.23.2
+go 1.25.0
 
 require (
 	contrib.go.opencensus.io/exporter/prometheus v0.4.2

media/ring.go

@@ -4,6 +4,7 @@ import (
 	"errors"
 	"io"
 	"sync"
+	"sync/atomic"
 )
 
 /*
@@ -213,12 +214,24 @@ func (rb *RingBuffer) MakeReader() *RingBufferReader {
 }
 
 type RingBufferReader struct {
-	rb *RingBuffer
-	nb int64
+	rb     *RingBuffer
+	nb     int64
+	closed atomic.Bool
 }
 
 func (rbr *RingBufferReader) Read(p []byte) (int, error) {
+	if closed := rbr.closed.Load(); closed {
+		return 0, io.EOF
+	}
 	n, err := rbr.rb.readFrom(p, rbr.nb)
 	rbr.nb += int64(n)
 	return n, err
 }
+
+func (rbr *RingBufferReader) Close() error {
+	// NB: If already blocking in Read(),
+	// EOF occurs on the *next* Read() call
+	// This does not wake up blocked readers.
+	rbr.closed.Store(true)
+	return nil
+}

media/ring_test.go

@@ -7,6 +7,7 @@ import (
 	"io"
 	"sync"
 	"testing"
+	"testing/synctest"
 	"time"
 
 	"github.com/stretchr/testify/assert"
@@ -415,3 +416,60 @@ func TestRingbuffer_WraparoundReads(t *testing.T) {
 	assert.Nil(err)
 	assert.Equal("67890", string(buf5))
 }
+
+func sync_TestRingbuffer_ReadClose(t *testing.T) {
+	assert := assert.New(t)
+
+	rbc := &RingBufferConfig{BufferLen: 8}
+	rb, err := NewRingBuffer(rbc)
+	assert.Nil(err)
+
+	reader := rb.MakeReader()
+	reader2 := rb.MakeReader()
+	var wg sync.WaitGroup
+
+	wg.Go(func() {
+		buf := make([]byte, 5)
+		n, err := reader.Read(buf)
+		assert.Equal(4, n)
+		assert.Nil(err)
+		assert.Equal([]byte{1, 2, 3, 4, 0}, buf)
+		n, err = reader.Read(buf)
+		assert.Equal(0, n)
+		assert.Equal(io.EOF, err)
+	})
+
+	wg.Go(func() {
+		// reader should be drained
+		buf := make([]byte, 5)
+		n, err := reader2.Read(buf)
+		assert.Equal(4, n)
+		assert.Nil(err)
+		assert.Equal([]byte{1, 2, 3, 4, 0}, buf)
+		n, err = reader2.Read(buf)
+		assert.Equal([]byte{5, 6, 7, 8, 0}, buf)
+		assert.Equal(4, n)
+		n, err = reader2.Read(buf)
+		assert.Equal(0, n)
+		assert.Equal(io.EOF, err)
+	})
+
+	// give the reader goroutine a moment to block waiting for data
+	time.Sleep(5 * time.Millisecond)
+
+	// close the reader concurrently
+	err = reader.Close()
+	assert.Nil(err)
+
+	rb.Write([]byte{1, 2, 3, 4})
+	time.Sleep(1 * time.Millisecond)
+	rb.Write([]byte{5, 6, 7, 8})
+
+	rb.Close()
+
+	wgWait(&wg)
+}
+
+func TestRingbuffer_ReadClose(t *testing.T) {
+	synctest.Test(t, sync_TestRingbuffer_ReadClose)
+}

media/rtp_timestamper.go

@@ -0,0 +1,37 @@
+package media
+
+const (
+	// MPEG‑TS PTS is 33 bits wide; it wraps at 2^33.
+	mpegtsMaxPts = 1 << 33
+	halfPts      = mpegtsMaxPts / 2
+)
+
+// RTPTimestamper keeps track of wrap‑arounds in a 33‑bit PTS stream.
+type RTPTimestamper struct {
+	lastRaw     int64 // last raw PTS seen
+	wrapCount   int64 // how many times we've wrapped
+	initialized bool  // false until first Unwrap call
+}
+
+// NewRTPTimestamper creates a fresh unwrapper.
+func NewRTPTimestamper() *RTPTimestamper {
+	return &RTPTimestamper{}
+}
+
+// returns a monotonic 32‑bit timeline by accounting for 33-bit wrap‑arounds.
+func (u *RTPTimestamper) ToTS(rawPts int64) uint32 {
+	if u.initialized {
+		delta := rawPts - u.lastRaw
+		if delta < -halfPts {
+			// wrapped forward
+			u.wrapCount++
+		} else if delta > halfPts {
+			// wrapped backward (unlikely in normal streams)
+			u.wrapCount--
+		}
+	} else {
+		u.initialized = true
+	}
+	u.lastRaw = rawPts
+	return uint32(rawPts + u.wrapCount*mpegtsMaxPts)
+}

media/rtp_timestamper_test.go

@@ -0,0 +1,96 @@
+package media
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+const (
+	mod32 = 1 << 32 // 4 294 967 296
+)
+
+func TestRTPTimestamper(t *testing.T) {
+	tests := []struct {
+		name    string
+		inputs  []int64
+		wantRtp []uint32
+	}{
+		{
+			name:    "Initialization",
+			inputs:  []int64{0},
+			wantRtp: []uint32{0},
+		},
+		{
+			name:    "Monotonic increase",
+			inputs:  []int64{0, 1, 2, 1000},
+			wantRtp: []uint32{0, 1, 2, 1000},
+		},
+		{
+			name:    "Small backwards jump (no wrap)",
+			inputs:  []int64{1000, 990},
+			wantRtp: []uint32{1000, 990},
+		},
+		{
+			name:   "Forward 33-bit PTS wrap",
+			inputs: []int64{mpegtsMaxPts - 2, 2},
+			wantRtp: []uint32{
+				uint32((mpegtsMaxPts - 2) % mod32),
+				uint32((2 + mpegtsMaxPts) % mod32),
+			},
+		},
+		{
+			name:   "Backward 33-bit PTS wrap",
+			inputs: []int64{5, mpegtsMaxPts - 3},
+			wantRtp: []uint32{
+				5,
+				uint32(mod32 - 3),
+			},
+		},
+		{
+			name:    "Crossing 32-bit boundary (wrap in RTP)",
+			inputs:  []int64{halfPts - 1, halfPts, halfPts + 1},
+			wantRtp: []uint32{uint32(halfPts - 1), 0, 1},
+		},
+		{
+			name:   "Random jitter near wrap threshold",
+			inputs: []int64{mpegtsMaxPts - halfPts/2, 2},
+			wantRtp: []uint32{
+				uint32((mpegtsMaxPts - halfPts/2) % mod32),
+				2,
+			},
+		},
+		{
+			name:    "Large monotonic advance (< halfPts)",
+			inputs:  []int64{100, halfPts - 1},
+			wantRtp: []uint32{100, uint32(halfPts - 1)},
+		},
+		{
+			name:   "Multiple successive wraps",
+			inputs: []int64{mpegtsMaxPts - 1, 2, mpegtsMaxPts - 1, 2},
+			wantRtp: []uint32{
+				uint32((mpegtsMaxPts - 1) % mod32),
+				uint32((2 + mpegtsMaxPts) % mod32),
+				uint32((mpegtsMaxPts - 1) % mod32),
+				uint32((2 + mpegtsMaxPts) % mod32),
+			},
+		},
+		{
+			name:    "Idempotent Unwrap (same raw twice)",
+			inputs:  []int64{500, 500},
+			wantRtp: []uint32{500, 500},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			u := NewRTPTimestamper()
+			gotRtp := make([]uint32, len(tc.inputs))
+			for i, raw := range tc.inputs {
+				ext := u.ToTS(raw)
+				gotRtp[i] = ext
+			}
+			require.Equal(t, tc.wantRtp, gotRtp)
+		})
+	}
+}