From fb65801094007f2fe34a9e4d0f37a4a19b5fe166 Mon Sep 17 00:00:00 2001 From: Maurus Cuelenaere Date: Tue, 12 May 2026 22:31:34 +0200 Subject: [PATCH 1/4] feat(video): refcount the capture pipeline; expose pauseVideo / resumeVideo MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Introduces acquireVideoStream / releaseVideoStream / videoStreamHasConsumers helpers in video.go that gate the native encoder's VideoStart / VideoStop and the HDMI sleep ticker on a set of named consumers. The encoder runs iff at least one consumer holds it; the 0→1 transition runs VideoStart so the first delivered frame is an IDR. Each WebRTC Session gets a stable id (uuid in newSession) and its own consumer key "webrtc:". The slot is acquired when ICE reaches Connected and released when it reaches Closed. pauseVideo / resumeVideo JSON-RPC notifications toggle the calling session's own slot, handled inline in onRPCMessage so the dispatcher's session reference is in scope (the generic reflection-based dispatch doesn't pass *Session). This data model fixes by construction the two race classes Cursor Bugbot flagged on earlier drafts of this PR: - Handover-while-paused: the new session's own acquire is what starts the encoder; no special-case "restart if the outgoing session was paused" code at the handover site. - Stale-session pause: a pauseVideo from the soon-to-close session releases only its own slot, not the new session's; no need for a session != currentSession gate. The 1s handover overlap can briefly deliver frames to a paused session's track if the other session is keeping the encoder alive; bounded by the existing peer-connection close delay and acceptable for a feature aimed at sustained idle bandwidth saving. The refcount helpers are byte-identical to those in PR #1447 (commit dec25e1), so that PR will merge cleanly on top. Co-Authored-By: Claude Opus 4.7 --- jsonrpc.go | 16 ++++++++++++++++ video.go | 52 ++++++++++++++++++++++++++++++++++++++++++++++++++-- webrtc.go | 43 ++++++++++++++++++++++++++++++++++++++----- 3 files changed, 104 insertions(+), 7 deletions(-) diff --git a/jsonrpc.go b/jsonrpc.go index 7a60656e1..ead788f02 100644 --- a/jsonrpc.go +++ b/jsonrpc.go @@ -128,6 +128,22 @@ func onRPCMessage(message webrtc.DataChannelMessage, session *Session) { scopedLogger.Trace().Msg("Received RPC request") t := time.Now() + // pauseVideo / resumeVideo are session-bound notifications: they + // toggle this session's slot in the video stream refcount (see + // video.go). Handled inline because the generic dispatcher doesn't + // pass *Session, and acting on currentSession instead of the + // receiving session would let a stale data channel mis-target the + // active one during the 1s handover overlap. Both helpers are + // idempotent so rapid pause/resume bursts are safe. + switch request.Method { + case "pauseVideo": + releaseVideoStream(session.videoConsumerKey()) + return + case "resumeVideo": + acquireVideoStream(session.videoConsumerKey()) + return + } + handler, ok := rpcHandlers[request.Method] if !ok { errorResponse := JSONRPCResponse{ diff --git a/video.go b/video.go index e981979d6..677e44334 100644 --- a/video.go +++ b/video.go @@ -13,8 +13,56 @@ var ( lastVideoState native.VideoState videoSleepModeCtx context.Context videoSleepModeCancel context.CancelFunc + + videoConsumersMu sync.Mutex + videoConsumers = map[string]struct{}{} ) +// acquireVideoStream registers a named consumer of the capture pipeline. +// The first acquirer starts the native video stream and pauses the HDMI +// sleep ticker; subsequent acquirers from different consumers are recorded +// without touching the underlying stream. Idempotent per consumer key — +// re-acquiring an already-held key is a no-op. +func acquireVideoStream(consumer string) { + videoConsumersMu.Lock() + defer videoConsumersMu.Unlock() + + if _, exists := videoConsumers[consumer]; exists { + return + } + videoConsumers[consumer] = struct{}{} + if len(videoConsumers) == 1 { + _ = nativeInstance.VideoStart() + stopVideoSleepModeTicker() + } +} + +// releaseVideoStream unregisters a consumer. When the last consumer is +// released, the native video stream is stopped and the HDMI sleep ticker +// is restarted. Idempotent — releasing an unknown key is a no-op. +func releaseVideoStream(consumer string) { + videoConsumersMu.Lock() + defer videoConsumersMu.Unlock() + + if _, exists := videoConsumers[consumer]; !exists { + return + } + delete(videoConsumers, consumer) + if len(videoConsumers) == 0 { + _ = nativeInstance.VideoStop() + startVideoSleepModeTicker() + } +} + +// videoStreamHasConsumers reports whether any consumer currently holds the +// capture pipeline open. The HDMI sleep ticker uses this to decide whether +// it is safe to put the capture chip to sleep. +func videoStreamHasConsumers() bool { + videoConsumersMu.Lock() + defer videoConsumersMu.Unlock() + return len(videoConsumers) > 0 +} + const ( defaultVideoSleepModeDuration = 1 * time.Minute ) @@ -184,8 +232,8 @@ func doVideoSleepModeTicker(ctx context.Context, duration time.Duration) { for { select { case <-timer.C: - if getActiveSessions() > 0 { - nativeLogger.Warn().Msg("not going to enter HDMI sleep mode because there are active sessions") + if videoStreamHasConsumers() { + nativeLogger.Warn().Msg("not going to enter HDMI sleep mode because the capture pipeline has consumers") continue } diff --git a/webrtc.go b/webrtc.go index 7acd5f1d9..4172784fe 100644 --- a/webrtc.go +++ b/webrtc.go @@ -18,6 +18,7 @@ import ( "github.com/coder/websocket" "github.com/coder/websocket/wsjson" "github.com/gin-gonic/gin" + "github.com/google/uuid" "github.com/pion/ice/v4" "github.com/pion/interceptor" "github.com/pion/webrtc/v4" @@ -25,6 +26,11 @@ import ( ) type Session struct { + // id is a stable per-session identifier used as the suffix of the + // "webrtc:" consumer key for the video stream refcount in video.go. + // Generated in newSession; never reused. + id string + peerConnection *webrtc.PeerConnection VideoTrack *webrtc.TrackLocalStaticSample AudioTrack *webrtc.TrackLocalStaticSample @@ -48,6 +54,13 @@ type Session struct { codecMimeType string } +// videoConsumerKey is the key this session holds in the video stream +// refcount (see video.go). Acquired when ICE reaches Connected, released +// on Closed; pauseVideo / resumeVideo toggle it for this session only. +func (s *Session) videoConsumerKey() string { + return "webrtc:" + s.id +} + var ( actionSessions int = 0 activeSessionsMutex = &sync.Mutex{} @@ -509,6 +522,7 @@ func newSession(config SessionConfig) (*Session, error) { } session := &Session{ + id: uuid.New().String(), peerConnection: peerConnection, done: make(chan struct{}), rpcQueue: make(chan webrtc.DataChannelMessage, 256), @@ -604,7 +618,16 @@ func newSession(config SessionConfig) (*Session, error) { if incrActiveSessions() == 1 { onFirstSessionConnected() } + // Per-session setup (codec, audio start) must run + // before acquireVideoStream — the 0→1 refcount + // transition starts the encoder, which has to use + // this session's codec. onSessionConnected(session) + // Per-session slot in the video stream refcount; the + // 0→1 transition (first consumer overall) starts the + // encoder, so the first frame after a fresh acquire + // is an IDR. + acquireVideoStream(session.videoConsumerKey()) if mqttManager != nil { mqttManager.publishSessionsState() } @@ -642,6 +665,10 @@ func newSession(config SessionConfig) (*Session, error) { } if isConnected { isConnected = false + // Drop our slot in the video stream refcount. Idempotent — + // if pauseVideo already released it the call is a no-op. + // On the N→0 transition the encoder is stopped. + releaseVideoStream(session.videoConsumerKey()) onActiveSessionsChanged() if decrActiveSessions() == 0 { scopedLogger.Info().Msg("last session disconnected, stopping video stream") @@ -665,16 +692,22 @@ func onActiveSessionsChanged() { // capture is a shared pipeline; starting it again on a handoff connect (count // 1→2) would issue redundant native start calls and re-run the sleep-mode // re-lock wait while video is already streaming. +// +// VideoStart / sleep-ticker stop are owned by the video stream refcount +// (acquireVideoStream in video.go); we only handle the global lifecycle +// concerns that aren't a refcount transition here. func onFirstSessionConnected() { - stopVideoSleepModeTicker() _ = setHostDisplayAdvertised(true, "first_session_connected", false) - _ = nativeInstance.VideoStart() } // onSessionConnected runs per session when ICE reaches Connected. Uses the // session parameter directly rather than the currentSession global — that // global is assigned by the caller AFTER ExchangeOffer returns, and ICE // connected can fire before then, racing the assignment. +// +// Codec selection must happen here (before the per-session +// acquireVideoStream that follows) so the encoder restarts with this +// session's codec on the refcount 0→1 transition. func onSessionConnected(session *Session) { notifyFailsafeMode(session) if session.codecMimeType == webrtc.MimeTypeH265 { @@ -688,10 +721,10 @@ func onSessionConnected(session *Session) { } func onLastSessionDisconnected() { - // Safety net: ensure all keys are released when the last session disconnects + // Safety net: ensure all keys are released when the last session disconnects. + // VideoStop / sleep ticker are owned by releaseVideoStream (called when each + // session's ICE state reaches Closed, just above this). _ = rpcKeyboardReport(0, keyboardClearStateKeys) stopAudio() - _ = nativeInstance.VideoStop() _ = applyHostDisplayAdvertisement("last_session_disconnected") - startVideoSleepModeTicker() } From 72f3bce02736747c00cdda34d5893330cf248905 Mon Sep 17 00:00:00 2001 From: Maurus Cuelenaere Date: Mon, 11 May 2026 17:56:18 +0200 Subject: [PATCH 2/4] feat(ui): pause video stream when tab is hidden via Page Visibility API Mirrors the new server-side pauseVideo/resumeVideo JSON-RPC methods in the web frontend. When the user switches tabs or minimizes the browser, the encoder feed stops and outbound RTP drops to keepalive levels until the tab regains visibility. State is synced on mount so a reconnect into an already-hidden tab pauses immediately. Co-Authored-By: Claude Opus 4.7 --- ui/src/components/WebRTCVideo.tsx | 24 ++++++++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/ui/src/components/WebRTCVideo.tsx b/ui/src/components/WebRTCVideo.tsx index 36a4be61f..5aedd6c5b 100644 --- a/ui/src/components/WebRTCVideo.tsx +++ b/ui/src/components/WebRTCVideo.tsx @@ -546,6 +546,30 @@ export default function WebRTCVideo({ [keyDownHandler, keyUpHandler, resetKeyboardState], ); + // Pause/resume the server-side video feed when the tab is hidden so we + // don't burn WAN bandwidth decoding-then-discarding frames the user + // can't see. The encoder is restarted on resume so the first frame is + // an IDR and decode is artifact-free. + useEffect( + function pauseVideoOnTabHidden() { + const sync = () => { + sendRpc(document.hidden ? "pauseVideo" : "resumeVideo", {}); + }; + + // Sync once on mount in case the tab is already hidden when we + // (re)connect, then track every visibility change. + sync(); + + const abortController = new AbortController(); + document.addEventListener("visibilitychange", sync, { + signal: abortController.signal, + }); + + return () => abortController.abort(); + }, + [sendRpc], + ); + // Setup Video Event Listeners useEffect( function setupVideoEventListeners() { From 0706a90ca26f9d5f98f3194437fd08a490cce7c5 Mon Sep 17 00:00:00 2001 From: Maurus Cuelenaere Date: Tue, 12 May 2026 23:22:46 +0200 Subject: [PATCH 3/4] refactor(jsonrpc): support session injection and synchronous handlers in dispatcher MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Extends RPCHandler with two new orthogonal flags: - TakesSession: dispatcher injects the receiving *Session as the handler's first reflected argument. Lets session-bound RPCs act on the session that delivered the message rather than the global currentSession. - Synchronous: handler runs inline on the per-session rpcQueue pump goroutine instead of in a fresh per-message goroutine. Preserves dequeue order for ordering-sensitive RPCs. Plumbs *Session through callRPCHandler / riskyCallRPCHandler and shifts the reflected param index by one when TakesSession is set. Splits the existing onRPCMessage into: - onRPCMessage: parse + lookup + sync/async decision; runs on the pump. - invokeRPCHandler: handler invocation + response write; runs either inline (Synchronous) or in a goroutine (default). The goroutine spawn previously living in the WebRTC pump (webrtc.go:381) moves into onRPCMessage, where the JSON-RPC layer can decide per handler. Existing handlers all keep zero-value defaults (TakesSession: false, Synchronous: false), so their dispatch behaviour is identical: today's "go onRPCMessage(...)" is now "go invokeRPCHandler(...)" — one extra stack frame, otherwise the same. No handler uses the new flags in this commit; that follows. Co-Authored-By: Claude Opus 4.7 --- jsonrpc.go | 83 +++++++++++++++++++++++++++++++++++++++++------------- webrtc.go | 7 +++-- 2 files changed, 68 insertions(+), 22 deletions(-) diff --git a/jsonrpc.go b/jsonrpc.go index ead788f02..6621ef2e4 100644 --- a/jsonrpc.go +++ b/jsonrpc.go @@ -99,6 +99,12 @@ func writeJSONRPCEvent(event string, params any, session *Session) { } } +// onRPCMessage parses a single RPC message off the per-session pump and +// dispatches it. Runs synchronously on the pump goroutine; handlers +// flagged Synchronous keep running there (so dequeue order is preserved +// for ordering-sensitive RPCs like pauseVideo / resumeVideo), everything +// else is dispatched in a fresh goroutine so a slow handler can't +// head-of-line block the queue. func onRPCMessage(message webrtc.DataChannelMessage, session *Session) { var request JSONRPCRequest err := json.Unmarshal(message.Data, &request) @@ -120,14 +126,6 @@ func onRPCMessage(message webrtc.DataChannelMessage, session *Session) { return } - scopedLogger := jsonRpcLogger.With(). - Str("method", request.Method). - Interface("params", request.Params). - Interface("id", request.ID).Logger() - - scopedLogger.Trace().Msg("Received RPC request") - t := time.Now() - // pauseVideo / resumeVideo are session-bound notifications: they // toggle this session's slot in the video stream refcount (see // video.go). Handled inline because the generic dispatcher doesn't @@ -158,7 +156,26 @@ func onRPCMessage(message webrtc.DataChannelMessage, session *Session) { return } - result, err := callRPCHandler(scopedLogger, handler, request.Params) + if handler.Synchronous { + invokeRPCHandler(request, handler, session) + } else { + go invokeRPCHandler(request, handler, session) + } +} + +// invokeRPCHandler runs a single RPC handler and writes its response back +// to the session. Called either inline on the pump (Synchronous handlers) +// or from a per-message goroutine (the default). +func invokeRPCHandler(request JSONRPCRequest, handler RPCHandler, session *Session) { + scopedLogger := jsonRpcLogger.With(). + Str("method", request.Method). + Interface("params", request.Params). + Interface("id", request.ID).Logger() + + scopedLogger.Trace().Msg("Received RPC request") + t := time.Now() + + result, err := callRPCHandler(scopedLogger, handler, session, request.Params) if err != nil { scopedLogger.Error().Err(err).Msg("Error calling RPC handler") errorResponse := JSONRPCResponse{ @@ -540,10 +557,25 @@ type RPCHandler struct { Func any Params []string OptionalParams []string + + // TakesSession: the handler's first parameter is *Session and the + // dispatcher injects the receiving session into it. Used by + // session-bound RPCs (e.g. pauseVideo / resumeVideo) that must act + // on the session whose data channel delivered the message, not on + // the global currentSession. + TakesSession bool + + // Synchronous: the handler runs inline on the per-session rpcQueue + // pump goroutine rather than in a fresh per-message goroutine. Use + // for ordering-sensitive handlers — pause/resume toggle a shared + // global refcount and the Go scheduler can otherwise reorder a + // tight pause→resume pair. Slow or blocking handlers MUST stay + // async (the default). + Synchronous bool } // call the handler but recover from a panic to ensure our RPC thread doesn't collapse on malformed calls -func callRPCHandler(logger zerolog.Logger, handler RPCHandler, params map[string]any) (result any, err error) { +func callRPCHandler(logger zerolog.Logger, handler RPCHandler, session *Session, params map[string]any) (result any, err error) { // Use defer to recover from a panic defer func() { if r := recover(); r != nil { @@ -557,11 +589,11 @@ func callRPCHandler(logger zerolog.Logger, handler RPCHandler, params map[string }() // Call the handler - result, err = riskyCallRPCHandler(logger, handler, params) + result, err = riskyCallRPCHandler(logger, handler, session, params) return result, err // do not combine these two lines into one, as it breaks the above defer function's setting of err } -func riskyCallRPCHandler(logger zerolog.Logger, handler RPCHandler, params map[string]any) (any, error) { +func riskyCallRPCHandler(logger zerolog.Logger, handler RPCHandler, session *Session, params map[string]any) (any, error) { handlerValue := reflect.ValueOf(handler.Func) handlerType := handlerValue.Type() @@ -572,8 +604,12 @@ func riskyCallRPCHandler(logger zerolog.Logger, handler RPCHandler, params map[s numParams := handlerType.NumIn() allParamNames := append(handler.Params, handler.OptionalParams...) //nolint:gocritic - if len(allParamNames) != numParams { - err := fmt.Errorf("mismatch between handler parameters (%d) and defined parameter names (%d)", numParams, len(allParamNames)) + expectedSlots := len(allParamNames) + if handler.TakesSession { + expectedSlots++ + } + if expectedSlots != numParams { + err := fmt.Errorf("mismatch between handler parameters (%d) and defined parameter names (%d)", numParams, expectedSlots) logger.Error().Strs("paramNames", allParamNames).Err(err).Msg("Cannot call RPC handler") return nil, err } @@ -584,14 +620,21 @@ func riskyCallRPCHandler(logger zerolog.Logger, handler RPCHandler, params map[s } args := make([]reflect.Value, numParams) + // Reflective params start after the injected *Session slot, if any. + argOffset := 0 + if handler.TakesSession { + args[0] = reflect.ValueOf(session) + argOffset = 1 + } - for i := range numParams { - paramType := handlerType.In(i) + for i := range len(allParamNames) { + paramType := handlerType.In(i + argOffset) paramName := allParamNames[i] paramValue, ok := params[paramName] + argIdx := i + argOffset if !ok { if optionalSet[paramName] { - args[i] = reflect.Zero(paramType) + args[argIdx] = reflect.Zero(paramType) continue } err := fmt.Errorf("missing parameter: %s", paramName) @@ -625,7 +668,7 @@ func riskyCallRPCHandler(logger zerolog.Logger, handler RPCHandler, params map[s newSlice.Index(j).Set(elemValue.Convert(paramType.Elem())) } } - args[i] = newSlice + args[argIdx] = newSlice } else if paramType.Kind() == reflect.Struct && convertedValue.Kind() == reflect.Map { jsonData, err := json.Marshal(convertedValue.Interface()) if err != nil { @@ -636,12 +679,12 @@ func riskyCallRPCHandler(logger zerolog.Logger, handler RPCHandler, params map[s if err := json.Unmarshal(jsonData, newStruct); err != nil { return nil, fmt.Errorf("failed to unmarshal JSON into struct: %v for parameter %s", err, paramName) } - args[i] = reflect.ValueOf(newStruct).Elem() + args[argIdx] = reflect.ValueOf(newStruct).Elem() } else { return nil, fmt.Errorf("invalid parameter type for: %s, type: %s", paramName, paramType.Kind()) } } else { - args[i] = convertedValue.Convert(paramType) + args[argIdx] = convertedValue.Convert(paramType) } } diff --git a/webrtc.go b/webrtc.go index 4172784fe..72228c36f 100644 --- a/webrtc.go +++ b/webrtc.go @@ -532,6 +532,10 @@ func newSession(config SessionConfig) (*Session, error) { rpcQueue := session.rpcQueue go func() { + // onRPCMessage runs synchronously on this pump goroutine so + // handlers flagged Synchronous (pause/resume) keep their + // dequeue order. The dispatcher spawns its own goroutine for + // every async handler internally. for { select { case <-session.done: @@ -543,8 +547,7 @@ func newSession(config SessionConfig) (*Session, error) { case <-session.done: return case msg := <-rpcQueue: - // TODO: only use goroutine if the task is asynchronous - go onRPCMessage(msg, session) + onRPCMessage(msg, session) } } }() From 20721570b9fd502e657060f108c2c8f1e94c973f Mon Sep 17 00:00:00 2001 From: Maurus Cuelenaere Date: Tue, 12 May 2026 23:29:23 +0200 Subject: [PATCH 4/4] fix(video): make pause/resume race-free by routing through the new dispatcher MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replaces the inline switch in onRPCMessage with a pair of regular RPCHandler entries flagged TakesSession + Synchronous. The dispatcher runs them on the per-session rpcQueue pump goroutine — which is the single sequential consumer of the queue — so a rapid pause→resume pair can no longer be reordered by the Go scheduler. Previously each RPC was dispatched via "go onRPCMessage(...)", letting the scheduler interleave the two helpers so a release could land after the matching acquire and leave the encoder stopped while the tab was visible. Reported by Cursor Bugbot: https://github.com/jetkvm/kvm/pull/1455#discussion_r3229730950 Co-Authored-By: Claude Opus 4.7 --- jsonrpc.go | 18 ++---------------- video.go | 16 ++++++++++++++++ 2 files changed, 18 insertions(+), 16 deletions(-) diff --git a/jsonrpc.go b/jsonrpc.go index 6621ef2e4..b9281bb7d 100644 --- a/jsonrpc.go +++ b/jsonrpc.go @@ -126,22 +126,6 @@ func onRPCMessage(message webrtc.DataChannelMessage, session *Session) { return } - // pauseVideo / resumeVideo are session-bound notifications: they - // toggle this session's slot in the video stream refcount (see - // video.go). Handled inline because the generic dispatcher doesn't - // pass *Session, and acting on currentSession instead of the - // receiving session would let a stale data channel mis-target the - // active one during the 1s handover overlap. Both helpers are - // idempotent so rapid pause/resume bursts are safe. - switch request.Method { - case "pauseVideo": - releaseVideoStream(session.videoConsumerKey()) - return - case "resumeVideo": - acquireVideoStream(session.videoConsumerKey()) - return - } - handler, ok := rpcHandlers[request.Method] if !ok { errorResponse := JSONRPCResponse{ @@ -1406,6 +1390,8 @@ var rpcHandlers = map[string]RPCHandler{ "wheelReport": {Func: rpcWheelReport, Params: []string{"wheelY", "wheelX"}}, "wakeHost": {Func: rpcWakeHost}, "getVideoState": {Func: rpcGetVideoState}, + "pauseVideo": {Func: rpcPauseVideo, TakesSession: true, Synchronous: true}, + "resumeVideo": {Func: rpcResumeVideo, TakesSession: true, Synchronous: true}, "getUSBState": {Func: rpcGetUSBState}, "unmountImage": {Func: rpcUnmountImage}, "rpcMountBuiltInImage": {Func: rpcMountBuiltInImage, Params: []string{"filename"}}, diff --git a/video.go b/video.go index 677e44334..083b5d91c 100644 --- a/video.go +++ b/video.go @@ -157,6 +157,22 @@ func updateHostDisplayAdvertisement(reason string, force bool) error { return setHostDisplayAdvertisedLocked(shouldAdvertiseHostDisplayLocked(), reason, force) } +// rpcPauseVideo releases this session's slot in the video stream +// refcount. Registered with TakesSession + Synchronous so it acts on +// the receiving session and can't be reordered relative to a +// resumeVideo from the same source. +func rpcPauseVideo(s *Session) error { + releaseVideoStream(s.videoConsumerKey()) + return nil +} + +// rpcResumeVideo re-acquires this session's slot. See rpcPauseVideo for +// the dispatcher flags this relies on. +func rpcResumeVideo(s *Session) error { + acquireVideoStream(s.videoConsumerKey()) + return nil +} + type rpcVideoSleepModeResponse struct { Supported bool `json:"supported"` Enabled bool `json:"enabled"`