fix(signaling): detect silent websocket partitions (#182)
Add a bounded authenticated control-plane watchdog that replaces only an unresponsive socket generation without granting it media-route authority.
This commit is contained in:
1 parent
f81577c4c5
commit
9080e68601
9 files changed
+623
-11
No files matched your search
@@ -479,10 +479,15 @@ viewers cannot attach behind any healthy root, a later reviewed config may
|
||||
admit additional direct server-fed edges under a separate explicit egress cap;
|
||||
it never creates a third endpoint edge or unbounded whole-room fanout.
|
||||
|
||||
Sub-second failure recovery is a target after failure detection. The current
|
||||
30-second control heartbeat cannot meet it for silent partitions, so the media
|
||||
plane needs a small 100-200 ms liveness/queue signal or an equivalent native
|
||||
transport event. This signal must be measured before its interval is fixed.
|
||||
Sub-second media recovery remains a target after media failure detection. A
|
||||
separate browser application challenge detects a silently partitioned signaling
|
||||
WebSocket on a bounded 5-second baseline plus two exact 2-second misses, then
|
||||
reauthenticates on a new socket without asking ordinary P2P or duplicate
|
||||
same-revision SFU media to close and without reporting `route-failed`. Existing
|
||||
`peer-selected` TURN authority remains session-bound and follows its bounded
|
||||
recovery path after session replacement. Control-plane recovery is not media
|
||||
evidence and does not replace a measured media liveness or native transport
|
||||
event for sub-second route recovery.
|
||||
|
||||
## Selected-Edge TURN After SFU
|
||||
|
||||
|
||||
@@ -491,10 +491,54 @@ media and broader audio/A-V verification remain open. The
|
||||
native host shared-encode sender is a separate planned phase regardless of this
|
||||
experiment's result and is not implemented here.
|
||||
|
||||
The recovery gate above does not cover silent network partitions. With the
|
||||
default 30-second heartbeat, server detection can take 30 to 60 seconds before
|
||||
the same 5-second grace begins; that case remains unverified and must be reported
|
||||
separately.
|
||||
## Browser Signaling Silent-Partition Detection
|
||||
|
||||
The recovery gate above does not cover a browser WebSocket that remains open
|
||||
locally after its network path has silently failed. The server's default
|
||||
30-second protocol heartbeat can take 30 to 60 seconds to observe that failure,
|
||||
and browser script cannot observe WebSocket Ping or Pong control frames. The
|
||||
[WHATWG WebSockets Living Standard](https://websockets.spec.whatwg.org/)
|
||||
(accessed 2026-08-22) exposes messages, generic error/close events, and
|
||||
`readyState`; it also specifies that constructing a WebSocket immediately
|
||||
starts a new connection while `close()` starts a closing handshake whose close
|
||||
event arrives later. The application therefore needs a small text-message
|
||||
challenge rather than waiting for either protocol Ping/Pong or the old socket's
|
||||
close event.
|
||||
|
||||
The retained compatibility seam is response-only opt-in under `screener-v5`:
|
||||
the Web client sends `signaling-challenge { sequence }`, and the server sends
|
||||
the exact `signaling-challenge-response { sequence }` only to that requesting
|
||||
socket. The Native sender's strict decoder rejects unknown server message
|
||||
types, so the server must never send this response unsolicited. Existing Web
|
||||
clients and Native senders do not opt in and therefore never receive the new
|
||||
message. A new server remains compatible with those old clients, but new Web
|
||||
assets and the signaling server still need an ordered, atomic deployment
|
||||
because an old server rejects the new client message. Supporting that reverse
|
||||
version skew would require explicit capability negotiation or a protocol bump.
|
||||
|
||||
The Web watchdog is eligible only while the current socket is authenticated,
|
||||
the share is online, and the endpoint has an active authoritative route (the
|
||||
Host itself, an ordinary Viewer with an online Host, or a peer-assisted Viewer
|
||||
whose active upstream is not `none`). After five seconds it sends one
|
||||
challenge. An exact two-second miss enters an internal suspect state and sends
|
||||
one fresh confirm challenge; an exact second two-second miss logically retires
|
||||
that socket generation and immediately constructs a replacement without
|
||||
waiting for the old closing handshake. Responses from another sequence or
|
||||
socket generation are ignored, and the server revalidates the current
|
||||
authenticated session before replying. A hidden document, a visibility
|
||||
restore, or an obviously late timer callback clears pending evidence and starts
|
||||
a fresh five-second baseline instead of declaring failure.
|
||||
|
||||
The response is socket-local, contains no room state, secret, candidate, or raw
|
||||
statistics, is not logged, and is limited by the server to at most one response
|
||||
per socket per second. The watchdog itself does not close or rebuild media,
|
||||
emit `route-failed`, or change route authority. Ordinary P2P and a duplicate
|
||||
same-revision SFU authority keep their healthy media while the new socket
|
||||
reauthenticates. An existing `peer-selected` TURN lease remains deliberately
|
||||
bound to both authenticated sessions: replacing either session invalidates that
|
||||
exceptional transport and leaves recovery to the existing bounded route path.
|
||||
Relaxing or carrying that security authority is a separate routing/protocol
|
||||
change, not part of signaling liveness detection.
|
||||
|
||||
## License Boundary
|
||||
|
||||
|
||||
@@ -142,6 +142,8 @@ Zod 在服务边界拒绝未知类型、超长字符串、非法 SDP/candidate
|
||||
|
||||
信令 WebSocket 暂时中断时,不立即关闭仍健康的 P2P 媒体;客户端指数退避恢复控制通道。明确不可重连的会话替换会释放旧页面的 peer、统计定时器和媒体引用,host 也停止旧捕获。首版只保证同一页面生命周期内恢复,不承诺浏览器进程被系统挂起后的连续播放。
|
||||
|
||||
Web 客户端只在当前 socket 已鉴权、分享在线且本端拥有 active 权威路由时运行应用层静默断链探测:五秒后发送带安全整数序号的 challenge,首个精确两秒 miss 只进入内部 suspect 并立即发送新序号 confirm,第二个精确两秒 miss 才废弃旧 socket 代次并直接建立新连接,不等待旧 closing handshake。response 必须匹配当前 socket 代次和序号;旧序号、旧 socket、已替换鉴权会话一律忽略。页面隐藏、恢复可见或定时器明显迟到时清除未决证据并重新等待完整五秒。服务端只向主动 opt-in 的请求 socket 回送相同序号,每 socket 每秒最多一次,不附带房间、密钥、候选或统计信息;因此 strict Native 不会收到未知消息。watchdog 本身不发送 `route-failed`、不修改路由权威,也不要求普通 P2P 或同 revision SFU 关闭健康媒体;现有 `peer-selected` TURN 授权仍严格绑定两端鉴权 session,任一 session 替换后按既有有界恢复路径处理。服务端 30 秒 ping/pong 默认值保持不变。
|
||||
|
||||
## 捕获、编码与质量
|
||||
|
||||
捕获只能由按钮产生的用户手势调用 `getDisplayMedia()`。三档目标为:
|
||||
|
||||
+213
-3
@@ -37,6 +37,15 @@ const INVALID_MESSAGE_CLOSE_CODE = 1008;
|
||||
const VIEWER_ACCESS_REVOKED_CLOSE_CODE = 4004;
|
||||
const PROTOCOL_REFRESH_MESSAGE = "页面版本已更新,请刷新后重试";
|
||||
const TERMINAL_SEND_TIMEOUT_MS = 15_000;
|
||||
const SIGNALING_CHALLENGE_INTERVAL_MS = 5_000;
|
||||
const SIGNALING_CHALLENGE_TIMEOUT_MS = 2_000;
|
||||
const SIGNALING_TIMER_LAG_TOLERANCE_MS = 1_000;
|
||||
|
||||
interface PendingSignalingChallenge {
|
||||
generation: number;
|
||||
sequence: number;
|
||||
confirm: boolean;
|
||||
}
|
||||
|
||||
export function shouldReconnectSignaling(code: number): boolean {
|
||||
return (
|
||||
@@ -61,16 +70,26 @@ export class SignalingClient {
|
||||
private authenticationTimer: number | null = null;
|
||||
private terminalTimer: number | null = null;
|
||||
private terminalMessage: ClientMessage | null = null;
|
||||
private socketGeneration = 0;
|
||||
private hostOnline = false;
|
||||
private authoritativeRoute = false;
|
||||
private challengeSequence = 0;
|
||||
private watchdogTimer: number | null = null;
|
||||
private watchdogDeadlineMs = 0;
|
||||
private pendingChallenge: PendingSignalingChallenge | null = null;
|
||||
private visibilityListenerAttached = false;
|
||||
|
||||
constructor(
|
||||
private readonly identity: SignalingIdentity,
|
||||
private readonly events: SignalingEvents,
|
||||
private readonly now: () => number = Date.now,
|
||||
) {}
|
||||
|
||||
start(): void {
|
||||
if (this.socket || this.stopped) {
|
||||
return;
|
||||
}
|
||||
this.attachVisibilityListener();
|
||||
this.events.onStatus(this.reconnectAttempt === 0 ? "connecting" : "reconnecting");
|
||||
this.connect();
|
||||
}
|
||||
@@ -79,6 +98,7 @@ export class SignalingClient {
|
||||
this.stopped = true;
|
||||
this.authenticated = false;
|
||||
this.clearTimers();
|
||||
this.detachVisibilityListener();
|
||||
this.terminalMessage = null;
|
||||
const socket = this.socket;
|
||||
this.socket = null;
|
||||
@@ -98,6 +118,7 @@ export class SignalingClient {
|
||||
) {
|
||||
return false;
|
||||
}
|
||||
this.clearSignalingWatchdog();
|
||||
socket.close(CLIENT_RECONNECT_CLOSE_CODE, "client reconnect");
|
||||
return true;
|
||||
}
|
||||
@@ -167,11 +188,16 @@ export class SignalingClient {
|
||||
return;
|
||||
}
|
||||
|
||||
const generation = ++this.socketGeneration;
|
||||
const socket = new WebSocket(signalUrl());
|
||||
this.socket = socket;
|
||||
|
||||
socket.addEventListener("open", () => {
|
||||
if (this.socket !== socket || this.stopped) {
|
||||
if (
|
||||
this.socket !== socket ||
|
||||
generation !== this.socketGeneration ||
|
||||
this.stopped
|
||||
) {
|
||||
return;
|
||||
}
|
||||
const authenticate: ClientMessage = {
|
||||
@@ -188,7 +214,11 @@ export class SignalingClient {
|
||||
});
|
||||
|
||||
socket.addEventListener("message", (event) => {
|
||||
if (this.socket !== socket || typeof event.data !== "string") {
|
||||
if (
|
||||
this.socket !== socket ||
|
||||
generation !== this.socketGeneration ||
|
||||
typeof event.data !== "string"
|
||||
) {
|
||||
return;
|
||||
}
|
||||
let message: ServerMessage;
|
||||
@@ -199,6 +229,11 @@ export class SignalingClient {
|
||||
return;
|
||||
}
|
||||
|
||||
if (message.type === "signaling-challenge-response") {
|
||||
this.acceptSignalingChallengeResponse(message.sequence, generation);
|
||||
return;
|
||||
}
|
||||
|
||||
if (
|
||||
message.type === "error" &&
|
||||
message.code === "INVALID_MESSAGE" &&
|
||||
@@ -231,15 +266,22 @@ export class SignalingClient {
|
||||
}
|
||||
this.events.onStatus("connected");
|
||||
}
|
||||
this.updateAuthoritativeActivity(message);
|
||||
this.events.onMessage(message);
|
||||
});
|
||||
|
||||
socket.addEventListener("close", (event) => {
|
||||
if (this.socket !== socket) {
|
||||
if (
|
||||
this.socket !== socket ||
|
||||
generation !== this.socketGeneration
|
||||
) {
|
||||
return;
|
||||
}
|
||||
this.socket = null;
|
||||
this.authenticated = false;
|
||||
this.hostOnline = false;
|
||||
this.authoritativeRoute = false;
|
||||
this.clearSignalingWatchdog();
|
||||
this.clearAuthenticationTimer();
|
||||
if (!this.stopped && shouldReconnectSignaling(event.code)) {
|
||||
this.scheduleReconnect();
|
||||
@@ -287,6 +329,7 @@ export class SignalingClient {
|
||||
|
||||
private clearTimers(): void {
|
||||
this.clearAuthenticationTimer();
|
||||
this.clearSignalingWatchdog();
|
||||
if (this.reconnectTimer !== null) {
|
||||
window.clearTimeout(this.reconnectTimer);
|
||||
this.reconnectTimer = null;
|
||||
@@ -296,4 +339,171 @@ export class SignalingClient {
|
||||
this.terminalTimer = null;
|
||||
}
|
||||
}
|
||||
|
||||
private updateAuthoritativeActivity(message: ServerMessage): void {
|
||||
if (message.type === "authenticated") {
|
||||
this.hostOnline = message.hostOnline;
|
||||
this.authoritativeRoute =
|
||||
this.identity.role === "host" ||
|
||||
!("mediaMode" in message) ||
|
||||
message.routeAssignment.upstream.kind !== "none";
|
||||
} else if (message.type === "host-status") {
|
||||
this.hostOnline = message.online;
|
||||
} else if (
|
||||
message.type === "route-update" &&
|
||||
message.phase === "active" &&
|
||||
this.identity.role === "viewer"
|
||||
) {
|
||||
this.authoritativeRoute = message.assignment.upstream.kind !== "none";
|
||||
} else if (
|
||||
message.type === "sharing-stopped" ||
|
||||
message.type === "room-closed"
|
||||
) {
|
||||
this.hostOnline = false;
|
||||
this.authoritativeRoute = false;
|
||||
} else {
|
||||
return;
|
||||
}
|
||||
this.refreshSignalingWatchdog();
|
||||
}
|
||||
|
||||
private refreshSignalingWatchdog(): void {
|
||||
if (!this.isWatchdogEligible()) {
|
||||
this.clearSignalingWatchdog();
|
||||
return;
|
||||
}
|
||||
if (this.watchdogTimer === null && this.pendingChallenge === null) {
|
||||
this.armSignalingWatchdog(SIGNALING_CHALLENGE_INTERVAL_MS);
|
||||
}
|
||||
}
|
||||
|
||||
private armSignalingWatchdog(delayMs: number): void {
|
||||
const generation = this.socketGeneration;
|
||||
this.watchdogDeadlineMs = this.now() + delayMs;
|
||||
this.watchdogTimer = window.setTimeout(() => {
|
||||
this.watchdogTimer = null;
|
||||
this.handleSignalingWatchdogTick(generation);
|
||||
}, delayMs);
|
||||
}
|
||||
|
||||
private handleSignalingWatchdogTick(generation: number): void {
|
||||
if (generation !== this.socketGeneration) {
|
||||
return;
|
||||
}
|
||||
if (
|
||||
!this.isWatchdogEligible() ||
|
||||
Math.abs(this.now() - this.watchdogDeadlineMs) >
|
||||
SIGNALING_TIMER_LAG_TOLERANCE_MS
|
||||
) {
|
||||
this.rebaselineSignalingWatchdog();
|
||||
return;
|
||||
}
|
||||
const pending = this.pendingChallenge;
|
||||
if (!pending) {
|
||||
this.sendSignalingChallenge(false, generation);
|
||||
} else if (pending.confirm) {
|
||||
this.replaceUnresponsiveSocket(generation);
|
||||
} else {
|
||||
this.sendSignalingChallenge(true, generation);
|
||||
}
|
||||
}
|
||||
|
||||
private sendSignalingChallenge(confirm: boolean, generation: number): void {
|
||||
const socket = this.socket;
|
||||
if (!socket || generation !== this.socketGeneration) {
|
||||
return;
|
||||
}
|
||||
this.challengeSequence =
|
||||
this.challengeSequence === Number.MAX_SAFE_INTEGER
|
||||
? 0
|
||||
: this.challengeSequence + 1;
|
||||
const sequence = this.challengeSequence;
|
||||
socket.send(JSON.stringify({ type: "signaling-challenge", sequence }));
|
||||
this.pendingChallenge = { generation, sequence, confirm };
|
||||
this.armSignalingWatchdog(SIGNALING_CHALLENGE_TIMEOUT_MS);
|
||||
}
|
||||
|
||||
private acceptSignalingChallengeResponse(
|
||||
sequence: number,
|
||||
generation: number,
|
||||
): void {
|
||||
const pending = this.pendingChallenge;
|
||||
if (
|
||||
!pending ||
|
||||
pending.generation !== generation ||
|
||||
pending.sequence !== sequence ||
|
||||
generation !== this.socketGeneration
|
||||
) {
|
||||
return;
|
||||
}
|
||||
this.clearSignalingWatchdog();
|
||||
this.refreshSignalingWatchdog();
|
||||
}
|
||||
|
||||
private replaceUnresponsiveSocket(generation: number): void {
|
||||
const socket = this.socket;
|
||||
if (!socket || generation !== this.socketGeneration || this.stopped) {
|
||||
return;
|
||||
}
|
||||
this.socket = null;
|
||||
this.authenticated = false;
|
||||
this.hostOnline = false;
|
||||
this.authoritativeRoute = false;
|
||||
this.clearAuthenticationTimer();
|
||||
this.clearSignalingWatchdog();
|
||||
this.events.onStatus("reconnecting");
|
||||
if (socket.readyState < WebSocket.CLOSING) {
|
||||
socket.close(CLIENT_RECONNECT_CLOSE_CODE, "signaling timeout");
|
||||
}
|
||||
this.connect();
|
||||
}
|
||||
|
||||
private rebaselineSignalingWatchdog(): void {
|
||||
this.clearSignalingWatchdog();
|
||||
this.refreshSignalingWatchdog();
|
||||
}
|
||||
|
||||
private clearSignalingWatchdog(): void {
|
||||
this.pendingChallenge = null;
|
||||
if (this.watchdogTimer !== null) {
|
||||
window.clearTimeout(this.watchdogTimer);
|
||||
this.watchdogTimer = null;
|
||||
}
|
||||
}
|
||||
|
||||
private readonly onVisibilityChange = (): void => {
|
||||
this.rebaselineSignalingWatchdog();
|
||||
};
|
||||
|
||||
private attachVisibilityListener(): void {
|
||||
if (typeof document === "undefined" || this.visibilityListenerAttached) {
|
||||
return;
|
||||
}
|
||||
document.addEventListener("visibilitychange", this.onVisibilityChange);
|
||||
this.visibilityListenerAttached = true;
|
||||
}
|
||||
|
||||
private detachVisibilityListener(): void {
|
||||
if (typeof document === "undefined" || !this.visibilityListenerAttached) {
|
||||
return;
|
||||
}
|
||||
document.removeEventListener("visibilitychange", this.onVisibilityChange);
|
||||
this.visibilityListenerAttached = false;
|
||||
}
|
||||
|
||||
private isDocumentVisible(): boolean {
|
||||
return (
|
||||
typeof document === "undefined" || document.visibilityState === "visible"
|
||||
);
|
||||
}
|
||||
|
||||
private isWatchdogEligible(): boolean {
|
||||
return (
|
||||
this.authenticated &&
|
||||
this.hostOnline &&
|
||||
this.authoritativeRoute &&
|
||||
this.isDocumentVisible() &&
|
||||
this.socket?.readyState === WebSocket.OPEN
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -38,6 +38,7 @@ type ErrorCode = Extract<ServerMessage, { type: "error" }>["code"];
|
||||
const MAX_BUFFERED_SIGNAL_BYTES = 256 * 1024;
|
||||
const DEFAULT_MAX_SIGNAL_CONNECTIONS = 2_048;
|
||||
const DEFAULT_MAX_UNAUTHENTICATED_CONNECTIONS = 256;
|
||||
const MIN_SIGNALING_CHALLENGE_INTERVAL_MS = 1_000;
|
||||
|
||||
interface AuthenticatedSession {
|
||||
roomId: string;
|
||||
@@ -58,6 +59,7 @@ interface SocketState {
|
||||
revoked?: boolean;
|
||||
authenticating?: boolean;
|
||||
viewerPasswordUpdatePending?: boolean;
|
||||
lastSignalingChallengeAtMs?: number;
|
||||
authenticated?: AuthenticatedSession;
|
||||
}
|
||||
|
||||
@@ -634,6 +636,29 @@ export class SignalingServer {
|
||||
message: Exclude<ClientMessage, { type: "authenticate" }>,
|
||||
): void {
|
||||
switch (message.type) {
|
||||
case "signaling-challenge": {
|
||||
const state = this.socketStates.get(socket);
|
||||
if (!state || state.authenticated !== authenticated) {
|
||||
return;
|
||||
}
|
||||
const now = this.now();
|
||||
if (state.lastSignalingChallengeAtMs !== undefined) {
|
||||
const elapsedMs = now - state.lastSignalingChallengeAtMs;
|
||||
if (elapsedMs < 0) {
|
||||
state.lastSignalingChallengeAtMs = now;
|
||||
return;
|
||||
}
|
||||
if (elapsedMs < MIN_SIGNALING_CHALLENGE_INTERVAL_MS) {
|
||||
return;
|
||||
}
|
||||
}
|
||||
state.lastSignalingChallengeAtMs = now;
|
||||
this.send(socket, {
|
||||
type: "signaling-challenge-response",
|
||||
sequence: message.sequence,
|
||||
});
|
||||
return;
|
||||
}
|
||||
case "signal":
|
||||
this.routeSignal(socket, authenticated, message);
|
||||
return;
|
||||
|
||||
@@ -536,8 +536,20 @@ const authenticateMessageSchema = z.discriminatedUnion("role", [
|
||||
.strict(),
|
||||
]);
|
||||
|
||||
const signalingChallengeSequenceSchema = z
|
||||
.number()
|
||||
.int()
|
||||
.min(0)
|
||||
.max(Number.MAX_SAFE_INTEGER);
|
||||
|
||||
export const clientMessageSchema = z.union([
|
||||
authenticateMessageSchema,
|
||||
z
|
||||
.object({
|
||||
type: z.literal("signaling-challenge"),
|
||||
sequence: signalingChallengeSequenceSchema,
|
||||
})
|
||||
.strict(),
|
||||
z
|
||||
.object({
|
||||
type: z.literal("signal"),
|
||||
@@ -680,6 +692,12 @@ const authenticatedMessageSchema = z.union([
|
||||
|
||||
export const serverMessageSchema = z.union([
|
||||
authenticatedMessageSchema,
|
||||
z
|
||||
.object({
|
||||
type: z.literal("signaling-challenge-response"),
|
||||
sequence: signalingChallengeSequenceSchema,
|
||||
})
|
||||
.strict(),
|
||||
z
|
||||
.object({
|
||||
type: z.literal("peer-joined"),
|
||||
|
||||
@@ -40,6 +40,7 @@ import {
|
||||
} from "../src/client/webrtc/stats.ts";
|
||||
|
||||
afterEach(() => {
|
||||
vi.useRealTimers();
|
||||
vi.unstubAllGlobals();
|
||||
});
|
||||
|
||||
@@ -659,6 +660,251 @@ describe("client signaling recovery policy", () => {
|
||||
expect(signal.reconnect()).toBe(false);
|
||||
});
|
||||
|
||||
it("confirms an exact silent partition before replacing the socket immediately", () => {
|
||||
vi.useFakeTimers();
|
||||
let now = 0;
|
||||
const sockets: FakeWebSocket[] = [];
|
||||
class FakeWebSocket extends EventTarget {
|
||||
static readonly OPEN = 1;
|
||||
static readonly CLOSING = 2;
|
||||
readyState = FakeWebSocket.OPEN;
|
||||
readonly send = vi.fn();
|
||||
readonly close = vi.fn((code?: number, reason?: string) => {
|
||||
void code;
|
||||
void reason;
|
||||
this.readyState = FakeWebSocket.CLOSING;
|
||||
});
|
||||
|
||||
constructor(readonly url: string) {
|
||||
super();
|
||||
sockets.push(this);
|
||||
}
|
||||
}
|
||||
vi.stubGlobal("WebSocket", FakeWebSocket);
|
||||
vi.stubGlobal("window", {
|
||||
location: new URL("https://share.test/r/123456789012"),
|
||||
setTimeout: globalThis.setTimeout,
|
||||
clearTimeout: globalThis.clearTimeout,
|
||||
});
|
||||
const statuses: string[] = [];
|
||||
const onMessage = vi.fn();
|
||||
const onTerminated = vi.fn();
|
||||
const signal = new SignalingClient(
|
||||
{
|
||||
roomId: "123456789012",
|
||||
role: "host",
|
||||
token: "h".repeat(43),
|
||||
clientId: "host-client",
|
||||
},
|
||||
{
|
||||
onMessage,
|
||||
onStatus: (status) => statuses.push(status),
|
||||
onTerminated,
|
||||
onAccessRequired: () => undefined,
|
||||
},
|
||||
() => now,
|
||||
);
|
||||
const receive = (socket: FakeWebSocket, value: object) => {
|
||||
const event = new Event("message");
|
||||
Object.defineProperty(event, "data", { value: JSON.stringify(value) });
|
||||
socket.dispatchEvent(event);
|
||||
};
|
||||
const authenticate = (socket: FakeWebSocket) => {
|
||||
socket.dispatchEvent(new Event("open"));
|
||||
receive(socket, {
|
||||
type: "authenticated",
|
||||
protocol: "screener-v5",
|
||||
role: "host",
|
||||
peerId: "host_12345678",
|
||||
roomExpiresAt: null,
|
||||
maxViewers: 8,
|
||||
hostOnline: true,
|
||||
connectionId: null,
|
||||
viewerPeerIds: [],
|
||||
iceConfig: { iceServers: [] },
|
||||
viewerPolicy: "private-link",
|
||||
viewerAuthorizationGeneration: "viewer_generation_12345678",
|
||||
});
|
||||
};
|
||||
const advance = (milliseconds: number) => {
|
||||
now += milliseconds;
|
||||
vi.advanceTimersByTime(milliseconds);
|
||||
};
|
||||
|
||||
signal.start();
|
||||
authenticate(sockets[0]!);
|
||||
advance(5_000);
|
||||
const first = JSON.parse(String(sockets[0]!.send.mock.calls.at(-1)![0]));
|
||||
expect(first).toEqual({ type: "signaling-challenge", sequence: 1 });
|
||||
|
||||
advance(2_000);
|
||||
const confirm = JSON.parse(String(sockets[0]!.send.mock.calls.at(-1)![0]));
|
||||
expect(confirm).toEqual({ type: "signaling-challenge", sequence: 2 });
|
||||
receive(sockets[0]!, {
|
||||
type: "signaling-challenge-response",
|
||||
sequence: first.sequence,
|
||||
});
|
||||
advance(2_000);
|
||||
|
||||
expect(sockets).toHaveLength(2);
|
||||
expect(sockets[0]!.close).toHaveBeenCalledWith(4002, "signaling timeout");
|
||||
expect(statuses.at(-1)).toBe("reconnecting");
|
||||
expect(onTerminated).not.toHaveBeenCalled();
|
||||
expect(
|
||||
sockets[0]!.send.mock.calls
|
||||
.map(([value]) => JSON.parse(String(value)).type)
|
||||
.includes("route-failed"),
|
||||
).toBe(false);
|
||||
|
||||
receive(sockets[0]!, {
|
||||
type: "signaling-challenge-response",
|
||||
sequence: confirm.sequence,
|
||||
});
|
||||
const oldClose = new Event("close");
|
||||
Object.defineProperties(oldClose, {
|
||||
code: { value: 4002 },
|
||||
reason: { value: "signaling timeout" },
|
||||
});
|
||||
sockets[0]!.dispatchEvent(oldClose);
|
||||
authenticate(sockets[1]!);
|
||||
advance(5_000);
|
||||
const healthy = JSON.parse(String(sockets[1]!.send.mock.calls.at(-1)![0]));
|
||||
receive(sockets[1]!, {
|
||||
type: "signaling-challenge-response",
|
||||
sequence: healthy.sequence,
|
||||
});
|
||||
advance(2_000);
|
||||
|
||||
expect(sockets).toHaveLength(2);
|
||||
expect(onMessage).toHaveBeenCalledTimes(2);
|
||||
signal.stop();
|
||||
});
|
||||
|
||||
it("rebases inactive, hidden, and obviously delayed challenge windows", () => {
|
||||
vi.useFakeTimers();
|
||||
let now = 0;
|
||||
let visibilityState: DocumentVisibilityState = "visible";
|
||||
const documentTarget = new EventTarget();
|
||||
Object.defineProperty(documentTarget, "visibilityState", {
|
||||
get: () => visibilityState,
|
||||
});
|
||||
const sockets: FakeWebSocket[] = [];
|
||||
class FakeWebSocket extends EventTarget {
|
||||
static readonly OPEN = 1;
|
||||
static readonly CLOSING = 2;
|
||||
readyState = FakeWebSocket.OPEN;
|
||||
readonly send = vi.fn();
|
||||
readonly close = vi.fn();
|
||||
|
||||
constructor(readonly url: string) {
|
||||
super();
|
||||
sockets.push(this);
|
||||
}
|
||||
}
|
||||
vi.stubGlobal("WebSocket", FakeWebSocket);
|
||||
vi.stubGlobal("document", documentTarget);
|
||||
vi.stubGlobal("window", {
|
||||
location: new URL("https://share.test/r/123456789012"),
|
||||
setTimeout: globalThis.setTimeout,
|
||||
clearTimeout: globalThis.clearTimeout,
|
||||
});
|
||||
const onMessage = vi.fn();
|
||||
const signal = new SignalingClient(
|
||||
{
|
||||
roomId: "123456789012",
|
||||
role: "viewer",
|
||||
clientId: "viewer-client",
|
||||
},
|
||||
{
|
||||
onMessage,
|
||||
onStatus: () => undefined,
|
||||
onTerminated: () => undefined,
|
||||
onAccessRequired: () => undefined,
|
||||
},
|
||||
() => now,
|
||||
);
|
||||
const receive = (value: object) => {
|
||||
const event = new Event("message");
|
||||
Object.defineProperty(event, "data", { value: JSON.stringify(value) });
|
||||
sockets[0]!.dispatchEvent(event);
|
||||
};
|
||||
const advance = (milliseconds: number) => {
|
||||
now += milliseconds;
|
||||
vi.advanceTimersByTime(milliseconds);
|
||||
};
|
||||
|
||||
signal.start();
|
||||
sockets[0]!.dispatchEvent(new Event("open"));
|
||||
receive({
|
||||
type: "authenticated",
|
||||
protocol: "screener-v5",
|
||||
role: "viewer",
|
||||
peerId: "viewer_12345678",
|
||||
roomExpiresAt: null,
|
||||
maxViewers: 8,
|
||||
hostOnline: true,
|
||||
connectionId: "connection_12345678",
|
||||
viewerPeerIds: [],
|
||||
iceConfig: { iceServers: [] },
|
||||
viewerPolicy: "private-link",
|
||||
viewerAuthorizationGeneration: "viewer_generation_12345678",
|
||||
mediaMode: "peer-assisted",
|
||||
mediaAssignment: { parentPeerId: null, childPeerIds: [] },
|
||||
routeRevision: 1,
|
||||
routeAssignment: {
|
||||
upstream: { kind: "none" },
|
||||
childPeerIds: [],
|
||||
sfuPublicationGeneration: null,
|
||||
},
|
||||
qualitySettings: {
|
||||
resolution: "1080p",
|
||||
maxFramerate: 30,
|
||||
maxBitrate: 5_000_000,
|
||||
degradationPreference: "balanced",
|
||||
videoCodec: "automatic",
|
||||
screenAudioQuality: "music",
|
||||
},
|
||||
});
|
||||
advance(5_000);
|
||||
expect(sockets[0]!.send).toHaveBeenCalledTimes(1);
|
||||
|
||||
receive({
|
||||
type: "route-update",
|
||||
revision: 2,
|
||||
phase: "active",
|
||||
assignment: {
|
||||
upstream: { kind: "peer", peerId: "host_12345678" },
|
||||
childPeerIds: [],
|
||||
sfuPublicationGeneration: null,
|
||||
},
|
||||
});
|
||||
advance(5_000);
|
||||
const first = JSON.parse(String(sockets[0]!.send.mock.calls.at(-1)![0]));
|
||||
expect(first.type).toBe("signaling-challenge");
|
||||
|
||||
visibilityState = "hidden";
|
||||
documentTarget.dispatchEvent(new Event("visibilitychange"));
|
||||
advance(20_000);
|
||||
expect(sockets).toHaveLength(1);
|
||||
expect(sockets[0]!.send).toHaveBeenCalledTimes(2);
|
||||
|
||||
visibilityState = "visible";
|
||||
documentTarget.dispatchEvent(new Event("visibilitychange"));
|
||||
advance(5_000);
|
||||
const visible = JSON.parse(String(sockets[0]!.send.mock.calls.at(-1)![0]));
|
||||
receive({
|
||||
type: "signaling-challenge-response",
|
||||
sequence: visible.sequence,
|
||||
});
|
||||
now += 7_000;
|
||||
vi.advanceTimersByTime(5_000);
|
||||
expect(sockets[0]!.send).toHaveBeenCalledTimes(3);
|
||||
advance(5_000);
|
||||
expect(sockets[0]!.send).toHaveBeenCalledTimes(4);
|
||||
expect(onMessage).toHaveBeenCalledTimes(2);
|
||||
signal.stop();
|
||||
});
|
||||
|
||||
it.each([
|
||||
JSON.stringify({
|
||||
type: "error",
|
||||
|
||||
@@ -90,6 +90,39 @@ describe("client signaling protocol", () => {
|
||||
);
|
||||
});
|
||||
|
||||
it("keeps signaling challenges strict and sequence-only", () => {
|
||||
expect(
|
||||
clientMessageSchema.safeParse({
|
||||
type: "signaling-challenge",
|
||||
sequence: Number.MAX_SAFE_INTEGER,
|
||||
}).success,
|
||||
).toBe(true);
|
||||
expect(
|
||||
serverMessageSchema.safeParse({
|
||||
type: "signaling-challenge-response",
|
||||
sequence: 0,
|
||||
}).success,
|
||||
).toBe(true);
|
||||
for (const invalid of [
|
||||
{ type: "signaling-challenge", sequence: -1 },
|
||||
{ type: "signaling-challenge", sequence: 1.5 },
|
||||
{
|
||||
type: "signaling-challenge",
|
||||
sequence: Number.MAX_SAFE_INTEGER + 1,
|
||||
},
|
||||
{ type: "signaling-challenge", sequence: 1, roomId },
|
||||
{ type: "signaling-challenge-response", sequence: "1" },
|
||||
{ type: "signaling-challenge-response", sequence: 1, sessionId: token },
|
||||
]) {
|
||||
expect(
|
||||
(invalid.type === "signaling-challenge"
|
||||
? clientMessageSchema
|
||||
: serverMessageSchema
|
||||
).safeParse(invalid).success,
|
||||
).toBe(false);
|
||||
}
|
||||
});
|
||||
|
||||
it("accepts only the fixed provisional Host lease", () => {
|
||||
expect(
|
||||
createRoomRequestSchema.parse({ viewerPolicy: "private-link" }),
|
||||
|
||||
@@ -1353,6 +1353,35 @@ async function closeClient(client: TestClient): Promise<void> {
|
||||
}
|
||||
|
||||
describe("WebSocket signaling", () => {
|
||||
it("answers only opted-in current sockets and rate-limits each socket", async () => {
|
||||
let now = Date.now();
|
||||
const harness = await startHarness({ now: () => now });
|
||||
const host = await openClient(harness.webSocketUrl);
|
||||
await authenticate(host, harness.room, "host", "challenge-host");
|
||||
|
||||
host.socket.send(
|
||||
JSON.stringify({ type: "signaling-challenge", sequence: 7 }),
|
||||
);
|
||||
expect(await host.inbox.next("signaling-challenge-response")).toEqual({
|
||||
type: "signaling-challenge-response",
|
||||
sequence: 7,
|
||||
});
|
||||
|
||||
host.socket.send(
|
||||
JSON.stringify({ type: "signaling-challenge", sequence: 8 }),
|
||||
);
|
||||
await host.inbox.expectNone(40);
|
||||
|
||||
now += 1_000;
|
||||
host.socket.send(
|
||||
JSON.stringify({ type: "signaling-challenge", sequence: 9 }),
|
||||
);
|
||||
expect(await host.inbox.next("signaling-challenge-response")).toEqual({
|
||||
type: "signaling-challenge-response",
|
||||
sequence: 9,
|
||||
});
|
||||
});
|
||||
|
||||
it("keeps the exact Native Host wire unchanged without a presence opt-in", async () => {
|
||||
const harness = await startHarness();
|
||||
const host = await openClient(harness.webSocketUrl);
|
||||
|
||||
Reference in new issue
Block a user