major remote desktop improvements

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

@@ -4,6 +4,8 @@ import (
 	"sync"
 	"sync/atomic"
 	"time"
+
+	"overlord-client/cmd/agent/h264util"
 )
 
 type Kind string
@@ -28,30 +30,64 @@ type writerEntry struct {
 	audio AudioWriter
 }
 
-type latestVideoWriter struct {
-	writer  VideoWriter
-	mu      sync.Mutex
-	pending []byte
-	dur     time.Duration
-	closed  bool
-	wake    chan struct{}
+const maxPendingVideoFrames = 6
+
+type queuedVideoFrame struct {
+	data []byte
+	dur  time.Duration
+	key  bool
 }
 
-func newLatestVideoWriter(writer VideoWriter) *latestVideoWriter {
-	queued := &latestVideoWriter{writer: writer, wake: make(chan struct{}, 1)}
+type latestVideoWriter struct {
+	writer        VideoWriter
+	kind          Kind
+	mu            sync.Mutex
+	pending       []queuedVideoFrame
+	needsKeyframe bool
+	closed        bool
+	wake          chan struct{}
+}
+
+func newLatestVideoWriter(kind Kind, writer VideoWriter) *latestVideoWriter {
+	queued := &latestVideoWriter{writer: writer, kind: kind, wake: make(chan struct{}, 1)}
 	go queued.run()
 	return queued
 }
 
 func (w *latestVideoWriter) enqueue(frame []byte, dur time.Duration) {
+	isKey := h264util.IsIDR(frame)
 	w.mu.Lock()
 	if w.closed {
 		w.mu.Unlock()
 		return
 	}
-	w.pending = frame
-	w.dur = dur
+	if w.needsKeyframe && !isKey {
+		w.mu.Unlock()
+		RequestKeyframe(w.kind)
+		return
+	}
+
+	requestKeyframe := false
+	if len(w.pending) >= maxPendingVideoFrames {
+		w.pending = nil
+		w.needsKeyframe = true
+		requestKeyframe = true
+		if !isKey {
+			w.mu.Unlock()
+			RequestKeyframe(w.kind)
+			return
+		}
+	}
+	if isKey && w.needsKeyframe {
+		w.pending = nil
+		w.needsKeyframe = false
+		requestKeyframe = false
+	}
+	w.pending = append(w.pending, queuedVideoFrame{data: frame, dur: dur, key: isKey})
 	w.mu.Unlock()
+	if requestKeyframe {
+		RequestKeyframe(w.kind)
+	}
 	select {
 	case w.wake <- struct{}{}:
 	default:
@@ -77,13 +113,26 @@ func (w *latestVideoWriter) run() {
 				w.mu.Unlock()
 				return
 			}
-			frame, dur := w.pending, w.dur
-			w.pending = nil
-			w.mu.Unlock()
-			if len(frame) == 0 {
+			if len(w.pending) == 0 {
+				w.mu.Unlock()
 				break
 			}
-			_ = w.writer.WriteH264(frame, dur)
+			frame := w.pending[0]
+			w.pending[0] = queuedVideoFrame{}
+			w.pending = w.pending[1:]
+			w.mu.Unlock()
+			err := w.writer.WriteH264(frame.data, frame.dur)
+			requestKeyframe := false
+			w.mu.Lock()
+			if err != nil {
+				w.pending = nil
+				w.needsKeyframe = true
+				requestKeyframe = true
+			}
+			w.mu.Unlock()
+			if requestKeyframe {
+				RequestKeyframe(w.kind)
+			}
 		}
 	}
 }
@@ -107,7 +156,7 @@ func registerVideoWriter(kind Kind, id string, w VideoWriter) {
 	if entry.video != nil {
 		entry.video.close()
 	}
-	entry.video = newLatestVideoWriter(w)
+	entry.video = newLatestVideoWriter(kind, w)
 	bucket[id] = entry
 	writersMu.Unlock()
 	RequestKeyframe(kind)

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

@@ -4,6 +4,8 @@ import (
 	"sync"
 	"testing"
 	"time"
+
+	"overlord-client/cmd/agent/h264util"
 )
 
 type blockingVideoWriter struct {
@@ -25,7 +27,15 @@ func (w *blockingVideoWriter) WriteH264(frame []byte, _ time.Duration) error {
 	return nil
 }
 
-func TestWriteH264DoesNotBlockAndKeepsLatestPendingFrame(t *testing.T) {
+func annexBDelta(value byte) []byte {
+	return []byte{0, 0, 0, 1, 0x41, value}
+}
+
+func annexBKey(value byte) []byte {
+	return []byte{0, 0, 0, 1, 0x67, 1, 0, 0, 1, 0x68, 1, 0, 0, 1, 0x65, value}
+}
+
+func TestWriteH264DoesNotBlockAndRequestsRecoveryAfterDroppingDelta(t *testing.T) {
 	kind := Kind("queue-test")
 	id := "slow-writer"
 	writer := &blockingVideoWriter{
@@ -34,8 +44,9 @@ func TestWriteH264DoesNotBlockAndKeepsLatestPendingFrame(t *testing.T) {
 	}
 	registerVideoWriter(kind, id, writer)
 	defer unregisterWriter(kind, id)
+	_ = ConsumeKeyframeRequest(kind)
 
-	if err := WriteH264(kind, []byte{1}, time.Millisecond); err != nil {
+	if err := WriteH264(kind, annexBDelta(1), time.Millisecond); err != nil {
 		t.Fatal(err)
 	}
 	select {
@@ -45,11 +56,16 @@ func TestWriteH264DoesNotBlockAndKeepsLatestPendingFrame(t *testing.T) {
 	}
 
 	started := time.Now()
-	_ = WriteH264(kind, []byte{2}, time.Millisecond)
-	_ = WriteH264(kind, []byte{3}, time.Millisecond)
+	for frame := 2; frame <= maxPendingVideoFrames+2; frame++ {
+		_ = WriteH264(kind, annexBDelta(byte(frame)), time.Millisecond)
+	}
 	if elapsed := time.Since(started); elapsed > 100*time.Millisecond {
 		t.Fatalf("capture path blocked for %s", elapsed)
 	}
+	if !ConsumeKeyframeRequest(kind) {
+		t.Fatal("dropping a predictive frame did not request a keyframe")
+	}
+	_ = WriteH264(kind, annexBKey(9), time.Millisecond)
 	close(writer.release)
 
 	deadline := time.Now().Add(time.Second)
@@ -58,14 +74,14 @@ func TestWriteH264DoesNotBlockAndKeepsLatestPendingFrame(t *testing.T) {
 		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)
+			if frames[0][len(frames[0])-1] != 1 || !h264util.IsIDR(frames[1]) {
+				t.Fatalf("expected first frame followed by recovery IDR, got %v", frames)
 			}
 			return
 		}
 		time.Sleep(5 * time.Millisecond)
 	}
-	t.Fatal("latest pending frame was not delivered")
+	t.Fatal("recovery keyframe was not delivered")
 }
 
 func TestKeyframeRequestsAreIsolatedByStreamKind(t *testing.T) {

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

@@ -47,7 +47,7 @@ func drainRTCP(sender *webrtc.RTPSender, kind Kind) {
 	}
 }
 
-const h264SDPFmtpLine = "level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=640034"
+const h264SDPFmtpLine = "level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=4d0034"
 
 var rtcpKeyframeLogOnce sync.Once
 var (