small optimizations

Overlord-Client/cmd/agent/webrtcpub/state.go

@@ -24,10 +24,70 @@ type AudioWriter interface {
 }
 
 type writerEntry struct {
-	video VideoWriter
+	video *latestVideoWriter
 	audio AudioWriter
 }
 
+type latestVideoWriter struct {
+	writer  VideoWriter
+	mu      sync.Mutex
+	pending []byte
+	dur     time.Duration
+	closed  bool
+	wake    chan struct{}
+}
+
+func newLatestVideoWriter(writer VideoWriter) *latestVideoWriter {
+	queued := &latestVideoWriter{writer: writer, wake: make(chan struct{}, 1)}
+	go queued.run()
+	return queued
+}
+
+func (w *latestVideoWriter) enqueue(frame []byte, dur time.Duration) {
+	w.mu.Lock()
+	if w.closed {
+		w.mu.Unlock()
+		return
+	}
+	w.pending = frame
+	w.dur = dur
+	w.mu.Unlock()
+	select {
+	case w.wake <- struct{}{}:
+	default:
+	}
+}
+
+func (w *latestVideoWriter) close() {
+	w.mu.Lock()
+	w.closed = true
+	w.pending = nil
+	w.mu.Unlock()
+	select {
+	case w.wake <- struct{}{}:
+	default:
+	}
+}
+
+func (w *latestVideoWriter) run() {
+	for range w.wake {
+		for {
+			w.mu.Lock()
+			if w.closed {
+				w.mu.Unlock()
+				return
+			}
+			frame, dur := w.pending, w.dur
+			w.pending = nil
+			w.mu.Unlock()
+			if len(frame) == 0 {
+				break
+			}
+			_ = w.writer.WriteH264(frame, dur)
+		}
+	}
+}
+
 var (
 	writersMu sync.RWMutex
 	writers   = map[string]map[string]writerEntry{} // kind → id → entry
@@ -44,7 +104,10 @@ func registerVideoWriter(kind Kind, id string, w VideoWriter) {
 		writers[string(kind)] = bucket
 	}
 	entry := bucket[id]
-	entry.video = w
+	if entry.video != nil {
+		entry.video.close()
+	}
+	entry.video = newLatestVideoWriter(w)
 	bucket[id] = entry
 	writersMu.Unlock()
 	RequestKeyframe()
@@ -72,6 +135,9 @@ func unregisterWriter(kind Kind, id string) {
 	}
 	writersMu.Lock()
 	if bucket, ok := writers[string(kind)]; ok {
+		if entry, exists := bucket[id]; exists && entry.video != nil {
+			entry.video.close()
+		}
 		delete(bucket, id)
 		if len(bucket) == 0 {
 			delete(writers, string(kind))
@@ -103,14 +169,19 @@ func WriteH264(kind Kind, nalu []byte, dur time.Duration) error {
 	if len(nalu) == 0 {
 		return nil
 	}
+	frame := append([]byte(nil), nalu...)
 	writersMu.RLock()
-	defer writersMu.RUnlock()
 	bucket := writers[string(kind)]
+	targets := make([]*latestVideoWriter, 0, len(bucket))
 	for _, w := range bucket {
 		if w.video != nil {
-			_ = w.video.WriteH264(nalu, dur)
+			targets = append(targets, w.video)
 		}
 	}
+	writersMu.RUnlock()
+	for _, target := range targets {
+		target.enqueue(frame, dur)
+	}
 	return nil
 }
 
@@ -119,13 +190,17 @@ func WriteAudio(kind Kind, pcm []int16) error {
 		return nil
 	}
 	writersMu.RLock()
-	defer writersMu.RUnlock()
 	bucket := writers[string(kind)]
+	targets := make([]AudioWriter, 0, len(bucket))
 	for _, w := range bucket {
 		if w.audio != nil {
-			_ = w.audio.WriteAudio(pcm)
+			targets = append(targets, w.audio)
 		}
 	}
+	writersMu.RUnlock()
+	for _, target := range targets {
+		_ = target.WriteAudio(pcm)
+	}
 	return nil
 }
 

Overlord-Client/cmd/agent/webrtcpub/state_test.go

@@ -0,0 +1,69 @@
+package webrtcpub
+
+import (
+	"sync"
+	"testing"
+	"time"
+)
+
+type blockingVideoWriter struct {
+	started chan struct{}
+	release chan struct{}
+	mu      sync.Mutex
+	frames  [][]byte
+}
+
+func (w *blockingVideoWriter) WriteH264(frame []byte, _ time.Duration) error {
+	w.mu.Lock()
+	w.frames = append(w.frames, append([]byte(nil), frame...))
+	count := len(w.frames)
+	w.mu.Unlock()
+	if count == 1 {
+		close(w.started)
+		<-w.release
+	}
+	return nil
+}
+
+func TestWriteH264DoesNotBlockAndKeepsLatestPendingFrame(t *testing.T) {
+	kind := Kind("queue-test")
+	id := "slow-writer"
+	writer := &blockingVideoWriter{
+		started: make(chan struct{}),
+		release: make(chan struct{}),
+	}
+	registerVideoWriter(kind, id, writer)
+	defer unregisterWriter(kind, id)
+
+	if err := WriteH264(kind, []byte{1}, time.Millisecond); err != nil {
+		t.Fatal(err)
+	}
+	select {
+	case <-writer.started:
+	case <-time.After(time.Second):
+		t.Fatal("writer did not receive first frame")
+	}
+
+	started := time.Now()
+	_ = WriteH264(kind, []byte{2}, time.Millisecond)
+	_ = WriteH264(kind, []byte{3}, time.Millisecond)
+	if elapsed := time.Since(started); elapsed > 100*time.Millisecond {
+		t.Fatalf("capture path blocked for %s", elapsed)
+	}
+	close(writer.release)
+
+	deadline := time.Now().Add(time.Second)
+	for time.Now().Before(deadline) {
+		writer.mu.Lock()
+		frames := append([][]byte(nil), writer.frames...)
+		writer.mu.Unlock()
+		if len(frames) >= 2 {
+			if frames[0][0] != 1 || frames[1][0] != 3 {
+				t.Fatalf("expected first and latest frames, got %v", frames)
+			}
+			return
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+	t.Fatal("latest pending frame was not delivered")
+}

Overlord-Client/cmd/agent/webrtcpub/whip_pion.go

@@ -16,6 +16,7 @@ import (
 	"os"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/pion/rtcp"
@@ -46,23 +47,36 @@ func drainRTCP(sender *webrtc.RTPSender) {
 const h264SDPFmtpLine = "level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=640034"
 
 var rtcpKeyframeLogOnce sync.Once
+var lastRTCPKeyframeRequest atomic.Int64
+
+const rtcpKeyframeMinInterval = 750 * time.Millisecond
 
 func handleRTCPKeyframeFeedback() {
 	if !honorRTCPKeyframes() {
 		rtcpKeyframeLogOnce.Do(func() {
-			log.Printf("webrtcpub: ignoring RTCP video keyframe requests by default; set OVERLORD_WEBRTC_RTCP_KEYFRAMES=true to honor them")
+			log.Printf("webrtcpub: RTCP video keyframe recovery disabled by OVERLORD_WEBRTC_RTCP_KEYFRAMES")
 		})
 		return
 	}
+	now := time.Now().UnixNano()
+	for {
+		last := lastRTCPKeyframeRequest.Load()
+		if last > 0 && time.Duration(now-last) < rtcpKeyframeMinInterval {
+			return
+		}
+		if lastRTCPKeyframeRequest.CompareAndSwap(last, now) {
+			break
+		}
+	}
 	RequestKeyframe()
 }
 
 func honorRTCPKeyframes() bool {
 	switch strings.ToLower(strings.TrimSpace(os.Getenv("OVERLORD_WEBRTC_RTCP_KEYFRAMES"))) {
-	case "1", "true", "yes", "on":
-		return true
-	default:
+	case "0", "false", "no", "off":
 		return false
+	default:
+		return true
 	}
 }