Feature: Multi-device screen wall on the Go server

Bring the Go web backend up to parity with the MFC server's screen wall.
A browser tab can now stream several devices over one WebSocket: the
single `watching` field becomes a per-connection watch set guarded by the
hub mutex, capped at 32 devices per socket. The challenge advertises
`caps.multiWatch` and `caps.keyframeRequest`, binary frames stay keyed by
the low 32 bits of the device ID, and connect/disconnect replies echo the
device id and request id so the wall can route concurrent responses.

Input commands (mouse, key, clipboard, rdp_reset) resolve their target
from the watch set: an explicit id must be a current subscription, and an
absent id is accepted only while exactly one device is watched. Legacy
frontends that do not send `multiWatch` keep replace-all semantics, so a
repeated device switch cannot grow the set without bound. Offline
devices are removed from every watch set, and teardown is computed from a
snapshot taken outside the hub lock to avoid re-entrant locking.

Add the `keyframe` command: it replays the cached IDR, then asks the
device to emit a fresh one, rate-limited to one request per device per
second. This exposed a Go-only corruption bug: the client answers the
request with TOKEN_KEYFRAME (134), which the dispatcher treated as a
no-op. That token only appears on forced IDRs because the client's GOP is
effectively infinite, so the dropped IDR left the browser decoding
P-frames against a reference chain it never received, producing
persistent garbling. Relay it as a keyframe instead. Also drop the
cached IDR when the resolution changes, mirroring the C++ server, so a
late joiner cannot receive an old-geometry keyframe.

Add tests for the watch-set transitions and for relaying TOKEN_KEYFRAME.

Co-Authored-By: deepseek-flash
This commit is contained in:
yuanyuanxianganddeepseek-flash committed 2026-10-18 11:51:28 +02:00
1 parent 8eab26c1f6
commit 1922f87dd2
7 files changed
+502 -80

No files matched your search

+66 -18
View File
@@ -52,8 +52,16 @@ type wsClient struct {
token string // set once authenticated
role string // mirrors session role after login
addr string // client address for logs
watching string // device ID this browser is currently streaming, "" when on the list
termWatching string // device ID for an open web terminal session, "" otherwise
// watchIDs is the set of devices this browser is streaming (the screen
// wall watches several at once). Empty while on the device list.
// Guarded by wsHub.mu.
watchIDs map[string]struct{}
// keyframeAt throttles keyframe requests to one per device per second.
// Unix seconds, guarded by wsHub.mu.
keyframeAt map[string]int64
}
// queue writes a JSON text frame onto the send buffer. Drops silently if the
@@ -85,6 +93,24 @@ func (c *wsClient) close() {
})
}
// challengeCaps advertises optional protocol features. It is appended to every
// challenge, including the rotation after a failed login, because the browser
// overwrites its capability state on each challenge. Old frontends ignore it.
const challengeCaps = `,"caps":{"multiWatch":true,"keyframeRequest":true}`
// maxWatchPerClient bounds the number of devices one WebSocket may stream at
// once. The wall tops out at 25 tiles; the slack covers a focus switch racing
// a teardown. This keeps the per-frame broadcast loop cheap for a misbehaving
// or buggy client.
const maxWatchPerClient = 32
// isWatching reports whether this connection subscribed to deviceID.
// Callers must hold wsHub.mu (read or write).
func (c *wsClient) isWatching(deviceID string) bool {
_, ok := c.watchIDs[deviceID]
return ok
}
// ----- ws hub: registry of all connected browsers -------------------------
type wsHub struct {
@@ -253,10 +279,19 @@ func (h *wsHub) OnDeviceOffline(id string) {
"cmd": "device_offline",
"id": id,
})
h.mu.RLock()
defer h.mu.RUnlock()
// Drop the id from every watch set as well: the device is gone, so this
// connection no longer holds a subscription for it. The browser does the
// same with its local refcount. Without this, a stale entry would keep
// countWatchers non-zero and later block the screen teardown.
h.mu.Lock()
defer h.mu.Unlock()
for c := range h.clients {
if c.watching == id && c.token != "" {
if !c.isWatching(id) {
continue
}
delete(c.watchIDs, id)
delete(c.keyframeAt, id)
if c.token != "" {
c.queue(msg)
}
}
@@ -269,12 +304,13 @@ func (h *wsHub) OnDeviceOffline(id string) {
func (h *wsHub) OnCursorChange(deviceID string, index byte) {
msg := mustJSON(map[string]any{
"cmd": "cursor",
"id": deviceID,
"index": index,
})
h.mu.RLock()
defer h.mu.RUnlock()
for c := range h.clients {
if c.watching == deviceID && c.token != "" {
if c.isWatching(deviceID) && c.token != "" {
c.queue(msg)
}
}
@@ -294,7 +330,7 @@ func (h *wsHub) OnResolutionChange(deviceID string, width, height int) {
h.mu.RLock()
defer h.mu.RUnlock()
for c := range h.clients {
if c.watching == deviceID && c.token != "" {
if c.isWatching(deviceID) && c.token != "" {
c.queue(msg)
}
}
@@ -311,7 +347,7 @@ func (h *wsHub) OnClipboardText(deviceID string, text string) {
h.mu.RLock()
defer h.mu.RUnlock()
for c := range h.clients {
if c.watching == deviceID && c.token != "" {
if c.isWatching(deviceID) && c.token != "" {
c.queue(msg)
}
}
@@ -325,7 +361,7 @@ func (h *wsHub) OnScreenFrame(deviceID string, packet []byte, _ bool) {
h.mu.RLock()
defer h.mu.RUnlock()
for c := range h.clients {
if c.watching == deviceID && c.token != "" {
if c.isWatching(deviceID) && c.token != "" {
c.queueBinary(packet)
}
}
@@ -424,12 +460,22 @@ func (h *wsHub) register(c *wsClient) {
func (h *wsHub) unregister(c *wsClient) {
h.mu.Lock()
delete(h.clients, c)
// Snapshot the watch set and clear it under the same lock. The teardown
// below must run OUTSIDE the lock: countWatchers takes h.mu.RLock and
// deferredCloseScreen takes h.mu.Lock.
watched := make([]string, 0, len(c.watchIDs))
for id := range c.watchIDs {
watched = append(watched, id)
}
c.watchIDs = nil
c.keyframeAt = nil
h.mu.Unlock()
// If this client was the last viewer of a device, tear down the screen
// session so the device stops encoding. Done OUTSIDE the lock so the
// hub's mutators can take their own locks without risk of recursion.
if c.watching != "" && h.countWatchers(c.watching) == 0 {
h.deferredCloseScreen(c.watching)
// session so the device stops encoding.
for _, id := range watched {
if h.countWatchers(id) == 0 {
h.deferredCloseScreen(id)
}
}
// Terminal sessions are single-viewer by design, so any open session
// belongs to this client. Tear it down so the next viewer doesn't
@@ -492,11 +538,13 @@ func (h *wsHub) serve(w http.ResponseWriter, r *http.Request) {
}
client := &wsClient{
conn: conn,
send: make(chan wsMsg, wsSendBuffer),
closed: make(chan struct{}),
nonce: nonce,
addr: clientIP(r, h.trustForwardedFor),
conn: conn,
send: make(chan wsMsg, wsSendBuffer),
closed: make(chan struct{}),
nonce: nonce,
addr: clientIP(r, h.trustForwardedFor),
watchIDs: make(map[string]struct{}),
keyframeAt: make(map[string]int64),
}
h.register(client)
defer h.unregister(client)
@@ -504,7 +552,7 @@ func (h *wsHub) serve(w http.ResponseWriter, r *http.Request) {
go h.writeLoop(client)
// Greet with a challenge nonce so the browser can compute the login response.
client.queue([]byte(`{"cmd":"challenge","nonce":"` + nonce + `"}`))
client.queue([]byte(`{"cmd":"challenge","nonce":"` + nonce + `"` + challengeCaps + `}`))
h.readLoop(client)
}