fix(test): fix all test bugs

internal/client/client.go

@@ -220,6 +220,7 @@ func (c *Client) bringUpLink(
 func smuxConfig() *smux.Config {
 	cfg := smux.DefaultConfig()
 	cfg.Version = 2
+	cfg.KeepAliveDisabled = true
 	cfg.MaxFrameSize = 32768
 	cfg.MaxReceiveBuffer = 16 * 1024 * 1024
 	cfg.MaxStreamBuffer = 1024 * 1024

internal/client/client_test.go

@@ -38,7 +38,7 @@ func TestSetupCipherRejectsBadInput(t *testing.T) {
 
 func TestSmuxConfig(t *testing.T) {
 	cfg := smuxConfig()
-	if cfg.Version != 2 || cfg.MaxFrameSize != 32768 || cfg.MaxReceiveBuffer != 16*1024*1024 {
+	if cfg.Version != 2 || !cfg.KeepAliveDisabled || cfg.MaxFrameSize != 32768 || cfg.MaxReceiveBuffer != 16*1024*1024 {
 		t.Fatalf("smuxConfig() = %+v", cfg)
 	}
 }

internal/provider/jazz/peer.go

@@ -412,8 +412,8 @@ func (p *Peer) handleSignaling(_ context.Context) {
 	for {
 		var msg map[string]any
 		if err := p.ws.ReadJSON(&msg); err != nil {
-			logger.Debugf("ws read error: %v", err)
 			if !p.closed.Load() {
+				logger.Debugf("ws read error: %v", err)
 				p.queueReconnect()
 			}
 			return
@@ -521,7 +521,7 @@ func (p *Peer) handleSubscriberOffer(payload map[string]any) {
 	_ = p.ws.WriteJSON(map[string]any{
 		keyRoomID:    p.roomInfo.RoomID,
 		keyEvent:     "media-in",
-		"groupId":   p.groupID,
+		"groupId":    p.groupID,
 		keyRequestID: uuid.New().String(),
 		keyPayload: map[string]any{
 			"method": "rtc:answer",
@@ -553,7 +553,7 @@ func (p *Peer) sendPublisherOffer() {
 	_ = p.ws.WriteJSON(map[string]any{
 		keyRoomID:    p.roomInfo.RoomID,
 		keyEvent:     "media-in",
-		"groupId":   p.groupID,
+		"groupId":    p.groupID,
 		keyRequestID: uuid.New().String(),
 		keyPayload: map[string]any{
 			"method": "rtc:offer",

internal/provider/telemost/peer.go

@@ -321,7 +321,7 @@ func (p *Peer) setupPeerConnections(config webrtc.Configuration) error {
 		}
 
 		logger.Infof("telemost remote video track: codec=%s stream=%s track=%s",
-				track.Codec().MimeType, track.StreamID(), track.ID())
+			track.Codec().MimeType, track.StreamID(), track.ID())
 
 		if cb := p.videoTrackHandler(); cb != nil {
 			cb(track, receiver)
@@ -477,15 +477,15 @@ func (p *Peer) sendHello() error {
 		keyUID: uuid.New().String(),
 		"hello": map[string]interface{}{
 			"participantMeta": map[string]interface{}{
-				"name":        p.name,
-				"role":        "SPEAKER",
+				"name":         p.name,
+				"role":         "SPEAKER",
 				keyDescription: "",
-				"sendAudio":   false,
-				"sendVideo":   p.hasLocalVideoTracks(),
+				"sendAudio":    false,
+				"sendVideo":    p.hasLocalVideoTracks(),
 			},
 			"participantAttributes": map[string]interface{}{
-				"name":        p.name,
-				"role":        "SPEAKER",
+				"name":         p.name,
+				"role":         "SPEAKER",
 				keyDescription: "",
 			},
 			"sendAudio":         false,
@@ -523,8 +523,8 @@ func (p *Peer) handleSignaling(ctx context.Context) {
 	for {
 		var msg map[string]interface{}
 		if err := p.ws.ReadJSON(&msg); err != nil {
-			logger.Debugf("ws read error: %v", err)
 			if !p.closed.Load() {
+				logger.Debugf("ws read error: %v", err)
 				p.queueReconnect()
 			}
 			return
@@ -624,7 +624,7 @@ func (p *Peer) handleSdpOffer(offer map[string]interface{}, uid string, sendPub
 		keyUID: uuid.New().String(),
 		"subscriberSdpAnswer": map[string]interface{}{
 			keyPcSeq: int(pcSeq),
-			"sdp":   answer.SDP,
+			"sdp":    answer.SDP,
 		},
 	})
 	p.wsMu.Unlock()
@@ -656,7 +656,7 @@ func (p *Peer) handleSdpOffer(offer map[string]interface{}, uid string, sendPub
 	_ = p.ws.WriteJSON(map[string]interface{}{
 		keyUID: uuid.New().String(),
 		"publisherSdpOffer": map[string]interface{}{
-			keyPcSeq:  1,
+			keyPcSeq: 1,
 			"sdp":    pubOffer.SDP,
 			"tracks": p.publisherTrackDescriptions(),
 		},
@@ -804,7 +804,7 @@ func (p *Peer) publisherTrackDescriptions() []map[string]interface{} {
 			"label":          track.ID(),
 			"codecs":         map[string]interface{}{},
 			"groupId":        1,
-			keyDescription:    "",
+			keyDescription:   "",
 		})
 	}
 
@@ -979,7 +979,7 @@ func (p *Peer) sendPong(uid string) {
 	defer p.wsMu.Unlock()
 
 	_ = p.ws.WriteJSON(map[string]interface{}{
-		keyUID:  uid,
+		keyUID: uid,
 		"pong": map[string]interface{}{},
 	})
 }
@@ -1146,7 +1146,7 @@ func (p *Peer) setupICEHandlers() {
 				"sdpMid":        init.SDPMid,
 				"sdpMlineIndex": init.SDPMLineIndex,
 				"target":        "SUBSCRIBER",
-				keyPcSeq:         1,
+				keyPcSeq:        1,
 			},
 		})
 		p.wsMu.Unlock()
@@ -1165,7 +1165,7 @@ func (p *Peer) setupICEHandlers() {
 				"sdpMid":        init.SDPMid,
 				"sdpMlineIndex": init.SDPMLineIndex,
 				"target":        "PUBLISHER",
-				keyPcSeq:         1,
+				keyPcSeq:        1,
 			},
 		})
 		p.wsMu.Unlock()
@@ -1181,7 +1181,7 @@ func (p *Peer) sendLeave(uid string) bool {
 	}
 
 	leave := map[string]interface{}{
-		keyUID:   uid,
+		keyUID:  uid,
 		"leave": map[string]interface{}{},
 	}
 
@@ -1204,21 +1204,10 @@ func (p *Peer) Close() error {
 		} else {
 			p.removeAckWaiter(leaveUID)
 		}
-		p.stopTelemetry()
 	}
 
 	closeSignal(p.closeCh)
-
-	done := make(chan struct{})
-	go func() {
-		p.wg.Wait()
-		close(done)
-	}()
-
-	select {
-	case <-done:
-	case <-time.After(2 * time.Second):
-	}
+	p.stopSession()
 
 	if p.dc != nil {
 		_ = p.dc.Close()
@@ -1238,6 +1227,17 @@ func (p *Peer) Close() error {
 		p.wsMu.Unlock()
 	}
 
+	done := make(chan struct{})
+	go func() {
+		p.wg.Wait()
+		close(done)
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+	}
+
 	return nil
 }
 
@@ -1283,7 +1283,7 @@ func (p *Peer) sendAppPing() bool {
 	defer p.wsMu.Unlock()
 	if p.ws != nil {
 		if err := p.ws.WriteJSON(map[string]interface{}{
-			keyUID:  uuid.New().String(),
+			keyUID: uuid.New().String(),
 			"ping": map[string]interface{}{},
 		}); err != nil {
 			logger.Debugf("app ping error: %v", err)

internal/provider/wbstream/peer.go

@@ -86,7 +86,7 @@ func (p *Peer) Connect(ctx context.Context) error {
 			},
 		},
 		OnDisconnected: func() {
-			if p.onEnded != nil {
+			if !p.closed.Load() && p.onEnded != nil {
 				p.onEnded("disconnected from livekit")
 			}
 		},

internal/server/server.go

@@ -155,6 +155,7 @@ func (s *Server) setupResolver() {
 func smuxConfig() *smux.Config {
 	cfg := smux.DefaultConfig()
 	cfg.Version = 2
+	cfg.KeepAliveDisabled = true
 	cfg.MaxFrameSize = 32768
 	cfg.MaxReceiveBuffer = 16 * 1024 * 1024
 	cfg.MaxStreamBuffer = 1024 * 1024
@@ -306,7 +307,7 @@ func (s *Server) serve(ctx context.Context) {
 				return
 			default:
 			}
-			logger.Infof("AcceptStream returned %v - reinstalling session", err)
+			logger.Debugf("AcceptStream returned %v - reinstalling session", err)
 			s.reinstallSession(sess)
 			continue
 		}

internal/server/server_test.go

@@ -40,7 +40,7 @@ func TestSetupCipherRejectsBadInput(t *testing.T) {
 
 func TestSmuxConfig(t *testing.T) {
 	cfg := smuxConfig()
-	if cfg.Version != 2 || cfg.MaxFrameSize != 32768 || cfg.MaxReceiveBuffer != 16*1024*1024 {
+	if cfg.Version != 2 || !cfg.KeepAliveDisabled || cfg.MaxFrameSize != 32768 || cfg.MaxReceiveBuffer != 16*1024*1024 {
 		t.Fatalf("smuxConfig() = %+v", cfg)
 	}
 }

internal/transport/seichannel/transport_test.go

@@ -3,6 +3,10 @@ package seichannel
 import (
 	"bytes"
 	"testing"
+
+	"github.com/pion/rtp"
+	"github.com/pion/rtp/codecs"
+	"github.com/pion/webrtc/v4/pkg/media/samplebuilder"
 )
 
 func TestSEIRoundTrip(t *testing.T) {
@@ -21,6 +25,43 @@ func TestSEIRoundTrip(t *testing.T) {
 	}
 }
 
+func TestSEIRoundTripThroughRTPPacketizerAndSampleBuilder(t *testing.T) {
+	payload := []byte("hello through rtp")
+	accessUnit := buildVideoAccessUnit(payload)
+
+	payloader := &codecs.H264Payloader{}
+	packets := payloader.Payload(1200, accessUnit)
+	if len(packets) == 0 {
+		t.Fatal("H264 payloader returned no packets")
+	}
+
+	sb := samplebuilder.New(128, &codecs.H264Packet{}, 90000)
+	for i, packetPayload := range packets {
+		sb.Push(&rtp.Packet{
+			Header: rtp.Header{
+				SequenceNumber: uint16(i + 1),
+				Timestamp:      1234,
+				Marker:         i == len(packets)-1,
+			},
+			Payload: packetPayload,
+		})
+	}
+	sb.Flush()
+
+	sample := sb.Pop()
+	if sample == nil {
+		t.Fatal("samplebuilder returned nil sample")
+	}
+
+	got, err := extractVideoPayloads(sample.Data)
+	if err != nil {
+		t.Fatalf("extractVideoPayloads(sample) error = %v", err)
+	}
+	if len(got) != 1 || !bytes.Equal(got[0], payload) {
+		t.Fatalf("RTP SEI payloads = %q, want %q", got, payload)
+	}
+}
+
 func TestTransportFrameRoundTrip(t *testing.T) {
 	encoded := encodeDataFrame(42, 0xdeadbeef, 1024, 1, 3, []byte("chunk"))
 	decoded, err := decodeTransportFrame(encoded)

internal/transport/vp8channel/transport.go

@@ -26,6 +26,7 @@ const (
 	outboundQueueSize     = 1024
 	inboundQueueSize      = 1024
 	canSendHighWatermark  = 90 // percent
+	maxWireFPS            = 120
 	keepaliveIdlePeriod   = 100 * time.Millisecond
 )
 
@@ -82,9 +83,9 @@ type streamTransport struct {
 	// every outgoing VP8 frame. peerEpoch tracks the last epoch we observed
 	// from the remote so we can detect their restart and reset locally.
 	bindingToken uint32
-	localEpoch uint32
-	peerEpoch  atomic.Uint32
-	hadPeer    atomic.Bool
+	localEpoch   uint32
+	peerEpoch    atomic.Uint32
+	hadPeer      atomic.Bool
 
 	kcp         *kcpRuntime
 	kcpMu       sync.RWMutex
@@ -310,14 +311,7 @@ func (p *streamTransport) Features() transport.Features {
 func (p *streamTransport) writerLoop() {
 	defer close(p.writerDone)
 
-	// Send each sample at the wire-level rate (fps * batchSize) instead of
-	// bursting batchSize samples per frame interval. Bursting makes RTP
-	// timestamps disagree with wall-clock arrival, which the SFU interprets
-	// as huge jitter and starts throttling the stream after a few seconds.
-	sampleInterval := p.frameInterval / time.Duration(p.batchSize)
-	if sampleInterval <= 0 {
-		sampleInterval = p.frameInterval
-	}
+	sampleInterval := p.sampleInterval()
 
 	ticker := time.NewTicker(sampleInterval)
 	defer ticker.Stop()
@@ -355,6 +349,18 @@ func (p *streamTransport) writerLoop() {
 	}
 }
 
+func (p *streamTransport) sampleInterval() time.Duration {
+	sampleInterval := p.frameInterval
+	if p.batchSize > 1 {
+		sampleInterval = p.frameInterval / time.Duration(p.batchSize)
+	}
+	minInterval := time.Second / maxWireFPS
+	if sampleInterval < minInterval {
+		return minInterval
+	}
+	return sampleInterval
+}
+
 func (p *streamTransport) resetKCP() {
 	p.drainOutbound()
 	p.kcpMu.Lock()

internal/transport/vp8channel/transport_unit_test.go

@@ -6,6 +6,7 @@ import (
 	"encoding/binary"
 	"errors"
 	"testing"
+	"time"
 
 	"github.com/openlibrecommunity/olcrtc/internal/carrier"
 	"github.com/openlibrecommunity/olcrtc/internal/transport"
@@ -18,6 +19,21 @@ type fakeVideoSession struct {
 	err    error
 }
 
+func TestSampleIntervalIsCappedForLargeBatch(t *testing.T) {
+	tr := &streamTransport{
+		frameInterval: time.Second / 60,
+		batchSize:     64,
+	}
+	if got := tr.sampleInterval(); got != time.Second/maxWireFPS {
+		t.Fatalf("sampleInterval() = %v, want %v", got, time.Second/maxWireFPS)
+	}
+
+	tr.batchSize = 1
+	if got := tr.sampleInterval(); got != tr.frameInterval {
+		t.Fatalf("sampleInterval(batch=1) = %v, want %v", got, tr.frameInterval)
+	}
+}
+
 func (s *fakeVideoSession) Capabilities() carrier.Capabilities {
 	return carrier.Capabilities{VideoTrack: true}
 }