feat(video): refcount the capture pipeline; expose pauseVideo / resumeVideo

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:<id>". 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 <noreply@anthropic.com>

Author: mcuelenaere

jsonrpc.go

@@ -124,6 +124,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{

video.go

@@ -6,14 +6,63 @@ import (
 	"time"
 
 	"github.com/jetkvm/kvm/internal/native"
+	"github.com/jetkvm/kvm/internal/sync"
 )
 
 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
 )
@@ -111,8 +160,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
 			}
 

webrtc.go

@@ -17,12 +17,18 @@ 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/webrtc/v4"
 	"github.com/rs/zerolog"
 )
 
 type Session struct {
+	// id is a stable per-session identifier used as the suffix of the
+	// "webrtc:<id>" consumer key for the video stream refcount in video.go.
+	// Generated in newSession; never reused.
+	id string
+
 	peerConnection           *webrtc.PeerConnection
 	VideoTrack               *webrtc.TrackLocalStaticSample
 	ControlChannel           *webrtc.DataChannel
@@ -44,6 +50,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{}
@@ -354,7 +367,10 @@ func newSession(config SessionConfig) (*Session, error) {
 		return nil, err
 	}
 
-	session := &Session{peerConnection: peerConnection}
+	session := &Session{
+		id:             uuid.New().String(),
+		peerConnection: peerConnection,
+	}
 	session.rpcQueue = make(chan webrtc.DataChannelMessage, 256)
 	session.initQueues()
 	session.initKeysDownStateQueue()
@@ -435,6 +451,11 @@ func newSession(config SessionConfig) (*Session, error) {
 				if incrActiveSessions() == 1 {
 					onFirstSessionConnected()
 				}
+				// 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()
 				}
@@ -481,6 +502,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")
@@ -502,18 +527,20 @@ func onActiveSessionsChanged() {
 
 func onFirstSessionConnected() {
 	notifyFailsafeMode(currentSession)
+	// The video stream's start/stop lifecycle is owned by the refcount in
+	// video.go (acquireVideoStream / releaseVideoStream). What stays here
+	// is the global codec selection: it has to happen before the encoder
+	// is started by the first acquire that follows this function.
 	if currentSession != nil && currentSession.codecMimeType == webrtc.MimeTypeH265 {
 		_ = nativeInstance.VideoSetCodecType(1)
 	} else {
 		_ = nativeInstance.VideoSetCodecType(0)
 	}
-	_ = nativeInstance.VideoStart()
-	stopVideoSleepModeTicker()
 }
 
 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)
-	_ = nativeInstance.VideoStop()
-	startVideoSleepModeTicker()
 }