diff --git a/packages/server/__tests__/hub.test.ts b/packages/server/__tests__/hub.test.ts index 1d414be..4858463 100644 --- a/packages/server/__tests__/hub.test.ts +++ b/packages/server/__tests__/hub.test.ts @@ -1,5 +1,7 @@ import { fanout, + fanoutLossy, + FEED_BACKLOG_LIMIT_BYTES, OPEN_READY_STATE, selectRevokedSockets, sweepLiveness, @@ -42,6 +44,39 @@ describe('WebSocket hub helpers', () => { expect(healthy.sent).toEqual(['payload']); }); + it('skips a backed-up client on the state feed instead of queueing frames', () => { + // A client that cannot keep up used to accumulate a queue it had to drain + // before it could show the present, on a socket that never looked broken. + const behind = fakeSocket({ bufferedAmount: FEED_BACKLOG_LIMIT_BYTES + 1 }); + const keepingUp = fakeSocket({ bufferedAmount: 1_024 }); + const unknownBuffer = fakeSocket(); + const failed: HubSocket[] = []; + + const delivered = fanoutLossy([behind, keepingUp, unknownBuffer], 'frame', (s) => + failed.push(s) + ); + + expect(delivered).toBe(2); + expect(behind.sent).toEqual([]); + expect(keepingUp.sent).toEqual(['frame']); + expect(unknownBuffer.sent).toEqual(['frame']); + expect(failed).toEqual([]); + }); + + it('still isolates a throwing client on the state feed', () => { + const throwing = fakeSocket({ + bufferedAmount: 0, + send() { + throw new Error('gone'); + } + }); + const healthy = fakeSocket(); + const failed: HubSocket[] = []; + + expect(fanoutLossy([throwing, healthy], 'frame', (s) => failed.push(s))).toBe(1); + expect(failed).toEqual([throwing]); + }); + it('terminates missed peers and pings peers that answered', () => { const missed = fakeSocket(); const responsive = fakeSocket(); diff --git a/packages/server/src/hub.ts b/packages/server/src/hub.ts index a55c8e9..75daccc 100644 --- a/packages/server/src/hub.ts +++ b/packages/server/src/hub.ts @@ -8,6 +8,8 @@ export const WS_REASON_SESSION_REVOKED = 'session revoked'; export interface HubSocket { readyState: number; + /** Bytes the socket has accepted but not yet written to the network. */ + bufferedAmount?: number; send(payload: string): void; close(code?: number, reason?: string): void; terminate(): void; @@ -18,6 +20,14 @@ export interface LivenessState { alive: boolean; } +/** + * How much unwritten payload makes a socket "behind" for a superseded feed. + * Roughly a second of state frames for a large grid: enough that an ordinary + * hiccup rides through, small enough that a client can never accumulate a + * queue of stale frames it has to drain before showing the present. + */ +export const FEED_BACKLOG_LIMIT_BYTES = 512 * 1024; + /** Send to every open socket without letting one broken peer stop the fanout. */ export function fanout( sockets: Iterable, @@ -37,6 +47,35 @@ export function fanout( return delivered; } +/** + * Fanout for a feed where every message supersedes the last (grid state at + * 60fps): a client that cannot keep up skips frames instead of queueing them. + * + * Queueing is what freezes a UI while the rig keeps running — the socket stays + * healthy, so nothing reconnects, and the client renders a backlog that is + * already seconds old. Dropping frames means the next one it does get is the + * present. + */ +export function fanoutLossy( + sockets: Iterable, + payload: string, + onFailure: (socket: T, error: unknown) => void, + limitBytes = FEED_BACKLOG_LIMIT_BYTES +): number { + let delivered = 0; + for (const socket of sockets) { + if (socket.readyState !== OPEN_READY_STATE) continue; + if ((socket.bufferedAmount ?? 0) > limitBytes) continue; + try { + socket.send(payload); + delivered++; + } catch (error) { + onFailure(socket, error); + } + } + return delivered; +} + /** * Mark responsive sockets for the next sweep and terminate peers that missed * the previous ping. diff --git a/packages/server/src/server.ts b/packages/server/src/server.ts index d04dc3c..1cd136e 100644 --- a/packages/server/src/server.ts +++ b/packages/server/src/server.ts @@ -12,7 +12,7 @@ import { computeCoverage } from './coverage'; import type { BlendMode, CannonState, Orientation, Rotation } from './grid'; import {compositeLayer, createGrid, DEFAULT_ALPHA, defaultOrientation, mapUiToGrid, remapGridForUi, resetGrid, setAllTargets, setCannonTarget, shiftGrid, tickGrid } from './grid'; import { createHttpApp, lanVisitors, resolveUiDir } from './http-app'; -import { fanout, type LivenessState,selectRevokedSockets, sweepLiveness, WS_CLOSE_SESSION_REVOKED, WS_REASON_SESSION_REVOKED } from './hub'; +import { fanout, fanoutLossy, type LivenessState,selectRevokedSockets, sweepLiveness, WS_CLOSE_SESSION_REVOKED, WS_REASON_SESSION_REVOKED } from './hub'; import type { JwtPayload } from './jwt'; import { verifyJwt } from './jwt'; import { ServerPatternEngine } from './pattern-engine'; @@ -305,7 +305,10 @@ function broadcastState() { GRID_COLUMNS, GRID_ROWS, orientation ); const payload = JSON.stringify({ type: 'state', grid: output }); - fanout(wss.clients, payload, dropClient); + // Lossy: this goes out 60 times a second and each frame replaces the last, so + // a client that cannot keep up must skip frames rather than build a backlog + // that leaves it rendering the past while the rig runs on the present. + fanoutLossy(wss.clients, payload, dropClient); } function getCalibrationOutput(): CannonState[] { diff --git a/packages/ui/src/lib/use-socket.ts b/packages/ui/src/lib/use-socket.ts index 8f2cea6..de9256c 100644 --- a/packages/ui/src/lib/use-socket.ts +++ b/packages/ui/src/lib/use-socket.ts @@ -44,6 +44,8 @@ export function useSocket( let attempts = 0; let retry: ReturnType | null = null; let watchdog: ReturnType | null = null; + /** Closed on purpose because the page is hidden, not because it broke. */ + let suspended = false; const probe = async (path: string) => { const res = await fetch(path, { headers: { Authorization: `Bearer ${token}` } }); @@ -76,6 +78,7 @@ export function useSocket( }); // A revoked session can only reconnect into the same rejection, so stop // hammering the brain and let the app hand back the login screen. + if (suspended) return; if (!isSessionEndedCode(e.code)) retry = setTimeout(connect, retryDelay(attempts)); }; @@ -94,6 +97,28 @@ export function useSocket( }; connect(); + + // A hidden page cannot paint the 60fps state feed and its timers are + // throttled, so the frames pile up in the socket instead: it comes back to + // a backlog of stale state on a socket that never looked broken. Dropping + // the connection while hidden and taking a fresh one on return means what + // the operator sees is always the present. + const onVisibility = () => { + if (document.visibilityState === 'hidden') { + suspended = true; + if (retry) clearTimeout(retry); + retry = null; + wsRef.current?.close(); + wsRef.current = null; + return; + } + if (!suspended) return; + suspended = false; + attempts = 0; + if (!wsRef.current) connect(); + }; + document.addEventListener('visibilitychange', onVisibility); + watchdog = setInterval(() => { const ws = wsRef.current; if ( @@ -104,6 +129,7 @@ export function useSocket( return () => { disposed = true; + document.removeEventListener('visibilitychange', onVisibility); if (retry) clearTimeout(retry); if (watchdog) clearInterval(watchdog); wsRef.current?.close();