ai/live: Gateway native WHEP server

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)
+		})
+	}
+}