From 165fb83e60dc9735e64f986c30119018c42f1781 Mon Sep 17 00:00:00 2001 From: Ryan Bundy Date: Tue, 21 Jul 2026 06:25:39 -0400 Subject: [PATCH 1/6] feat: correlate room joins and presence events Add a correlated JOIN protocol and make SocketManager the single owner of the join/reconnect lifecycle (issue #69, stacked on the presence foundation). - join-protocol: each JOIN carries a joinId; ROOM_STATE/JOIN:error are matched on BOTH room and joinId so a stale/superseded response can never settle a newer attempt. joinRoomOverSocket now returns a cancelable JoinHandle. - reconnect-coordinator (new, dependency-free): owns a logical room intent that survives transport drops. A mid-join disconnect cancels the transport attempt but keeps the intent pending; reconnect starts a fresh correlated attempt with the same presenceId and resolves the original caller. A newer room supersedes the older intent; leaving/clearing cancels it. Exposes a replayable readiness phase so transport connectivity alone never implies room-ready. - socket.ts: thin adapter wiring instance connect/disconnect into the coordinator (v4: reconnect fires on the manager; connect re-fires on the instance). Defers the wire JOIN until connected and drops/volatile-emits room-tagged AV so nothing buffered offline flushes into a later room. - server.js: echo joinId on ROOM_STATE/JOIN:error; room-tag AV/AV:new/AV:del; guard every presence-mutating path on socket ownership; handle every repeated-JOIN transition deterministically (same/other room, changed presenceId, new-socket rebind); preserve transforms on an in-place rebind; reject AV whose room != the socket's current room; never throw on a malformed payload; validate joinId as a bounded string. - presence.ts: isPresenceEventForRoom rejects untagged/mismatched presence events. --- spa/server.js | 184 +++++++++++------- spa/src/join-protocol.ts | 124 ++++++++---- spa/src/presence.ts | 16 ++ spa/src/reconnect-coordinator.ts | 314 +++++++++++++++++++++++++++++++ spa/src/socket.ts | 133 ++++++++++--- 5 files changed, 643 insertions(+), 128 deletions(-) create mode 100644 spa/src/reconnect-coordinator.ts diff --git a/spa/server.js b/spa/server.js index 7ee0819a..f4c2e80a 100644 --- a/spa/server.js +++ b/spa/server.js @@ -81,52 +81,100 @@ io.on("connection", async function(socket) { socket.emit("VERSION", { version: package.version }); socket.on("JOIN", (data) => { + // Never throw on a malformed or missing payload - a bad client must + // not be able to crash the socket handler. + if (!data || typeof data !== "object") { + socket.emit("JOIN:error", { room: undefined, joinId: undefined, reason: "invalid_payload" }); + return; + } + const { room, joinId } = data; const tokenData = validJwt(data.token); if (!tokenData) { console.error("invalid token!"); - socket.emit("JOIN:error", { reason: "invalid_token" }); + socket.emit("JOIN:error", { room, joinId, reason: "invalid_token" }); return; } const presenceId = data.presenceId; - const MAX_PRESENCE_ID_LENGTH = 128; + const MAX_ID_LENGTH = 128; if ( typeof presenceId !== "string" || presenceId.length === 0 || - presenceId.length > MAX_PRESENCE_ID_LENGTH + presenceId.length > MAX_ID_LENGTH ) { console.error("JOIN has invalid presenceId!"); - socket.emit("JOIN:error", { reason: "invalid_presence_id" }); + socket.emit("JOIN:error", { room, joinId, reason: "invalid_presence_id" }); + return; + } + // The joinId correlates this attempt with its authoritative response so + // a stale/superseded reply can never settle a newer client attempt. + if (typeof joinId !== "string" || joinId.length === 0 || joinId.length > MAX_ID_LENGTH) { + console.error("JOIN has invalid joinId!"); + socket.emit("JOIN:error", { room, joinId: undefined, reason: "invalid_join_id" }); + return; + } + if (room === undefined || room === null || `${room}`.length === 0) { + socket.emit("JOIN:error", { room, joinId, reason: "invalid_room" }); return; } - const { room } = data; // memberId, username, and avatar are derived only from the verified // JWT - never from client-supplied data - so presenceId can be // freely client-chosen without letting a client impersonate another // account's identity. const memberId = tokenData.id; const key = presenceKey(memberId, presenceId); - const defaultPos = [0, 0, 0]; - const defaultRot = [0, 1, 0, 0]; const user = USERS.get(socket); - // Defensively leave any previously tracked room for this socket - // before joining the new one - guarantees one room per socket even - // if a client ever calls JOIN without a preceding unsubscribe. - if (user.room && user.room !== room) { - const oldPresence = user.presenceKey ? PRESENCE.get(user.presenceKey) : null; - socket.leave(user.room); - socket.to(user.room).emit("AV:del", { - id: socket.id, - memberId: oldPresence?.memberId, - presenceId: oldPresence?.presenceId, - username: user.username, - }); - if (user.presenceKey && oldPresence?.socketId === socket.id) { + // (A) Tear down a DIFFERENT logical presence this socket previously + // owned (e.g. the same socket re-JOINing with a different presenceId). + // Only if this socket still owns that record - never clobber a presence + // a newer socket now owns. + if (user.presenceKey && user.presenceKey !== key) { + const oldOwned = PRESENCE.get(user.presenceKey); + if (oldOwned && oldOwned.socketId === socket.id) { + socket.to(oldOwned.room).emit("AV:del", { + id: socket.id, + room: oldOwned.room, + memberId: oldOwned.memberId, + presenceId: oldOwned.presenceId, + username: oldOwned.username, + }); PRESENCE.delete(user.presenceKey); } } + // (B) The target logical presence (key). If a record for it already + // exists in a DIFFERENT room, the presence is relocating: announce its + // departure from the old room and drop the stale record so it re-enters + // the new room as a fresh presence. If it exists in the SAME room, this + // is a rebind (reconnect / redundant JOIN) - preserve its transform so a + // restarted socket server (or a reconnecting client) doesn't snap the + // avatar back to the origin, and don't re-announce it. + const existingForKey = PRESENCE.get(key); + let pos = [0, 0, 0]; + let rot = [0, 1, 0, 0]; + if (existingForKey) { + if (`${existingForKey.room}` === `${room}`) { + pos = existingForKey.pos; + rot = existingForKey.rot; + } else { + socket.to(existingForKey.room).emit("AV:del", { + id: existingForKey.socketId, + room: existingForKey.room, + memberId: existingForKey.memberId, + presenceId: existingForKey.presenceId, + username: existingForKey.username, + }); + PRESENCE.delete(key); + } + } + + // (C) This socket's own room membership: leave the prior room if the + // socket is moving to a different one. + if (user.room && `${user.room}` !== `${room}`) { + socket.leave(user.room); + } + const isNewPresence = !PRESENCE.has(key); user.avatar = tokenData.avatar; @@ -137,11 +185,11 @@ io.on("connection", async function(socket) { PRESENCE.set(key, { memberId, presenceId, - socketId: socket.id, + socketId: socket.id, // transport metadata - rebinds to the current socket username: tokenData.username, avatar: tokenData.avatar, - pos: defaultPos, - rot: defaultRot, + pos, + rot, room, }); @@ -150,15 +198,16 @@ io.on("connection", async function(socket) { // Give the joining client one authoritative snapshot of everyone // currently in the room (including itself) instead of an ad-hoc // AV:new/AV replay loop. Chat and the X_ITE avatar layer reconcile - // against this by logical presence key, independent of readiness. - socket.emit("ROOM_STATE", { room, presences: roomPresenceSnapshot(room) }); + // against this by logical presence key, independent of readiness. The + // joinId is echoed so the client can correlate it with its attempt. + socket.emit("ROOM_STATE", { room, joinId, presences: roomPresenceSnapshot(room) }); - // Only announce a genuinely new presence - repeating JOIN for the - // same room/presence (e.g. a redundant client call) must not spam - // peers with another "someone joined" broadcast. + // Only announce a genuinely new presence - a rebind/redundant JOIN for + // the same room/presence must not spam peers with "someone joined". if (isNewPresence) { socket.to(room).emit("AV:new", { id: socket.id, + room, memberId, presenceId, avatar: tokenData.avatar, @@ -175,28 +224,26 @@ io.on("connection", async function(socket) { //handle avatar related calls. socket.on("AV", function(msg) { - msg.id = socket.id; + if (!msg || typeof msg !== "object") return; const user = USERS.get(socket); - if (user?.presenceKey) { - const presence = PRESENCE.get(user.presenceKey); - if (presence) { - msg.memberId = presence.memberId; - msg.presenceId = presence.presenceId; - if (msg.pos) presence.pos = msg.pos; - if (msg.rot) presence.rot = msg.rot; - } - } - if (user?.room) { - socket.to(user.room).emit("AV", msg); - } - if (user) { - if (msg.pos) { - USERS.get(socket).pos = msg.pos; - } - if (msg.rot) { - USERS.get(socket).rot = msg.rot; - } - } + if (!user || !user.room) return; + const presence = user.presenceKey ? PRESENCE.get(user.presenceKey) : null; + // Only the socket that currently owns the logical presence may move or + // relay it - a stale/replaced socket must not broadcast under this key. + if (!presence || presence.socketId !== socket.id) return; + // Reject AV tagged for a room other than the socket's current + // authoritative room (e.g. an offline-buffered event flushed after a + // room change) so it can't mutate the new room. + if (msg.room !== undefined && `${msg.room}` !== `${user.room}`) return; + msg.id = socket.id; + msg.room = user.room; // authoritative room tag for the broadcast + msg.memberId = presence.memberId; + msg.presenceId = presence.presenceId; + if (msg.pos) presence.pos = msg.pos; + if (msg.rot) presence.rot = msg.rot; + socket.to(user.room).emit("AV", msg); + if (msg.pos) user.pos = msg.pos; + if (msg.rot) user.rot = msg.rot; }); //handle shared events @@ -280,13 +327,17 @@ io.on("connection", async function(socket) { const room = user.room; const presence = user.presenceKey ? PRESENCE.get(user.presenceKey) : null; socket.leave(room); - socket.to(room).emit("AV:del", { - id: socket.id, - memberId: presence?.memberId, - presenceId: presence?.presenceId, - username: user.username, - }); - if (user.presenceKey) { + // Only announce the departure and delete the record if this socket + // still owns the logical presence - a stale/replaced socket must not + // remove or announce a presence a newer socket now owns. + if (presence && presence.socketId === socket.id) { + socket.to(room).emit("AV:del", { + id: socket.id, + room, + memberId: presence.memberId, + presenceId: presence.presenceId, + username: user.username, + }); PRESENCE.delete(user.presenceKey); } // Clear so a later disconnect (without a rejoin in between) sees @@ -303,22 +354,19 @@ io.on("connection", async function(socket) { socket.on("disconnect", function() { const user = USERS.get(socket); const presence = user?.presenceKey ? PRESENCE.get(user.presenceKey) : null; - // Only announce a departure if this socket was still in a room. A - // socket that already unsubscribed has had user.room cleared (and its - // AV:del already sent there), so re-emitting here would broadcast to - // an undefined room - matches the guard used in the "AV" handler. - if (user?.room) { + // Announce the departure and remove the record only if this socket was + // still in a room AND still owns the logical presence. Guarding BOTH on + // socketId (not just the delete) means a stale/delayed disconnect from + // an old socket can neither delete nor broadcast AV:del for a presence a + // newer reconnected socket now owns. + if (user?.room && presence && presence.socketId === socket.id) { io.to(user.room).emit("AV:del", { id: socket.id, - memberId: presence?.memberId, - presenceId: presence?.presenceId, + room: user.room, + memberId: presence.memberId, + presenceId: presence.presenceId, username: user?.username, }); - } - // Only remove the presence record if this socket is still the one - // it's bound to - guards against a stale/delayed disconnect from an - // old socket clobbering a newer connection for the same presence. - if (user?.presenceKey && presence?.socketId === socket.id) { PRESENCE.delete(user.presenceKey); } USERS.delete(socket); diff --git a/spa/src/join-protocol.ts b/spa/src/join-protocol.ts index e2ce6b92..a624a13f 100644 --- a/spa/src/join-protocol.ts +++ b/spa/src/join-protocol.ts @@ -14,53 +14,101 @@ export interface EmitterLike { export const DEFAULT_JOIN_TIMEOUT_MS = 10000; +/** Upper bound on a client-supplied joinId, mirroring the server-side check. */ +export const MAX_JOIN_ID_LENGTH = 128; + +/** + * A single in-flight transport JOIN attempt. `promise` settles when the + * server confirms (ROOM_STATE) or rejects (JOIN:error/timeout); `cancel` + * lets the owner supersede or abandon the attempt (e.g. on a newer room + * request or a transport drop) - cancelling rejects `promise` with the + * given reason so the owner can distinguish it from a real failure. + */ +export interface JoinHandle { + promise: Promise; + cancel: (reason?: string) => void; +} + +/** True for a non-empty, length-bounded joinId string (client + server agree). */ +export function isValidJoinId(joinId: unknown): joinId is string { + return typeof joinId === "string" && joinId.length > 0 && joinId.length <= MAX_JOIN_ID_LENGTH; +} + /** * Emits JOIN and waits for the server's authoritative confirmation - either - * a ROOM_STATE for the room we asked to join, or an explicit JOIN:error. - * Resolving merely because JOIN was emitted (the previous behavior) let - * Chat/room-ready state flip to "ready" even on an invalid token or a - * dropped request; this makes readiness conditional on a real server - * response, with a timeout so a lost response can't hang forever. + * a ROOM_STATE for the room AND joinId we asked for, or an explicit + * JOIN:error for that same attempt. Correlating on both room and joinId + * means a stale response from a superseded attempt can never settle a newer + * one. Resolving merely because JOIN was emitted (the original behavior) let + * readiness flip to "ready" on an invalid token or dropped request; this + * makes readiness conditional on a real, correlated server response, with a + * timeout so a lost response can't hang a single attempt forever. + * + * Returns a {@link JoinHandle} rather than a bare promise so the caller can + * cancel the attempt without treating cancellation as a JOIN failure. */ export function joinRoomOverSocket( socket: EmitterLike, roomId: string | number, token: string, presenceId: string, + joinId: string, timeoutMs: number = DEFAULT_JOIN_TIMEOUT_MS, -): Promise { - return new Promise((resolve, reject) => { - let settled = false; - - const cleanup = () => { - socket.off("ROOM_STATE", onRoomState); - socket.off("JOIN:error", onError); - clearTimeout(timer); - }; - const onRoomState = (event: { room?: string | number }) => { - if (settled) return; - // A ROOM_STATE for a different room is a stale/unrelated response - // (e.g. from a join this one superseded) - not confirmation of ours. - if (`${event?.room}` !== `${roomId}`) return; - settled = true; - cleanup(); - resolve(); - }; - const onError = (event: { reason?: string }) => { - if (settled) return; - settled = true; - cleanup(); - reject(new Error(`JOIN failed: ${event?.reason || "unknown"}`)); - }; - const timer = setTimeout(() => { - if (settled) return; - settled = true; - cleanup(); - reject(new Error("JOIN timed out waiting for server confirmation")); - }, timeoutMs); +): JoinHandle { + let settled = false; + let resolveFn: () => void = () => undefined; + let rejectFn: (err: Error) => void = () => undefined; - socket.on("ROOM_STATE", onRoomState); - socket.on("JOIN:error", onError); - socket.emit("JOIN", { room: roomId, token, presenceId }); + const promise = new Promise((resolve, reject) => { + resolveFn = resolve; + rejectFn = reject; }); + + const cleanup = () => { + socket.off("ROOM_STATE", onRoomState); + socket.off("JOIN:error", onError); + clearTimeout(timer); + }; + const onRoomState = (event: { room?: string | number; joinId?: string }) => { + if (settled) return; + // Must match BOTH the room we asked for and this exact attempt's joinId. + // A ROOM_STATE for a different room or a different (older) attempt is a + // stale/unrelated response and must not confirm this one. + if (`${event?.room}` !== `${roomId}`) return; + if (event?.joinId !== joinId) return; + settled = true; + cleanup(); + resolveFn(); + }; + const onError = (event: { room?: string | number; joinId?: string; reason?: string }) => { + if (settled) return; + // Reject only the attempt this error correlates to (same room + joinId); + // a stale JOIN:error must not reject a newer attempt. + if (`${event?.room}` !== `${roomId}`) return; + if (event?.joinId !== joinId) return; + settled = true; + cleanup(); + rejectFn(new Error(`JOIN failed: ${event?.reason || "unknown"}`)); + }; + const timer = setTimeout(() => { + if (settled) return; + settled = true; + cleanup(); + rejectFn(new Error("JOIN timed out waiting for server confirmation")); + }, timeoutMs); + + const cancel = (reason = "superseded") => { + if (settled) return; + settled = true; + cleanup(); + // Reason is carried in the message so the owner can tell an intentional + // cancel (superseded / disconnected / cleared) from a real failure. + rejectFn(new Error(`JOIN cancelled: ${reason}`)); + }; + + socket.on("ROOM_STATE", onRoomState); + socket.on("JOIN:error", onError); + socket.emit("JOIN", { room: roomId, token, presenceId, joinId }); + + return { promise, cancel }; } diff --git a/spa/src/presence.ts b/spa/src/presence.ts index e9f60f00..630eaf4a 100644 --- a/spa/src/presence.ts +++ b/spa/src/presence.ts @@ -49,6 +49,22 @@ export function isSelfPresence( return theirKey === myKey; } +/** + * True only when a room-scoped presence event (AV / AV:new / AV:del) is tagged + * with, and matches, the room the client is currently in. A missing or + * mismatched room tag is rejected rather than accepted ambiguously - this is + * what stops an in-flight event from a room we've navigated away from (or a + * legacy untagged event) from mutating the current room's presence store. + */ +export function isPresenceEventForRoom( + eventRoom: string | number | null | undefined, + activeRoom: string | number | null | undefined, +): boolean { + if (eventRoom === null || eventRoom === undefined) return false; + if (activeRoom === null || activeRoom === undefined) return false; + return `${eventRoom}` === `${activeRoom}`; +} + export interface ReconcileResult { added: Presence[]; updated: Presence[]; diff --git a/spa/src/reconnect-coordinator.ts b/spa/src/reconnect-coordinator.ts new file mode 100644 index 00000000..cd55dd28 --- /dev/null +++ b/spa/src/reconnect-coordinator.ts @@ -0,0 +1,314 @@ +/** + * Owner of the room-join lifecycle, kept dependency-free (no `@/` imports, no + * socket.io-client import) so the Node test harness can drive the real state + * machine directly. `SocketManager` is a thin adapter that wires real socket + * events into this coordinator; all of the correlation, superseding, reconnect + * and readiness logic lives here. + * + * The central idea (issue #69) is a two-level model: + * + * - A *logical intent* - "I want to be in room R" - which survives transport + * interruptions. Its caller-facing promise resolves once ANY correlated + * attempt for R gets an authoritative ROOM_STATE, and is rejected only when + * the intent is superseded by a newer room, explicitly cleared, or fails + * definitively (JOIN:error / timeout). A mere transport drop never rejects + * it - the intent stays pending and is retried on reconnect. + * + * - Per-transport *attempts*, each a correlated `joinRoomOverSocket` call with + * its own joinId. Only the latest attempt for the current intent is live; + * older ones are invalidated so a stale settlement can never move state. + */ + +import { + EmitterLike, + JoinHandle, + joinRoomOverSocket, + DEFAULT_JOIN_TIMEOUT_MS, +} from "./join-protocol"; + +/** Replayable readiness phase - a late-subscribing consumer can read this + * instead of having missed the transition events. */ +export type LifecyclePhase = "idle" | "joining" | "ready" | "disconnected" | "failed"; + +/** Transition signals emitted to lifecycle subscribers (Chat input, avatar + * viewpoint recovery). `ready` is the first successful join; `resynced` is a + * successful join that recovered from a prior disconnect (so only it should + * surface a user-facing "reconnected" message). */ +export type LifecycleEvent = "ready" | "disconnected" | "resynced" | "failed"; + +export type LifecycleListener = (event: LifecycleEvent) => void; + +export interface CoordinatorDeps { + /** The live transport (real socket.io Socket, or a fake EmitterLike in tests). */ + socket: EmitterLike; + /** Stable per-tab presence id, reused across reconnects. */ + presenceId: string; + /** Probe for whether the transport is currently connected. */ + isConnected: () => boolean; + /** Mints a unique joinId per attempt (injected for deterministic tests). */ + generateJoinId: () => string; + /** Per-attempt confirmation timeout. */ + joinTimeoutMs?: number; + /** Optional debug sink (kept dependency-free - no `@/helpers` here). */ + debug?: (...args: any[]) => void; +} + +export class ReconnectCoordinator { + private readonly socket: EmitterLike; + private readonly presenceId: string; + private readonly isConnected: () => boolean; + private readonly generateJoinId: () => string; + private readonly joinTimeoutMs: number; + private readonly debug: (...args: any[]) => void; + + // --- logical intent --- + private desiredRoom: string | number | null = null; + private desiredToken: string | null = null; + /** Bumped whenever the logical intent changes (new room / cleared), so a + * settlement from a prior intent can be recognised as stale and ignored. */ + private intentGeneration = 0; + private logicalResolve: (() => void) | null = null; + private logicalReject: ((err: Error) => void) | null = null; + + // --- current transport attempt --- + private currentAttempt: JoinHandle | null = null; + private pendingJoinIdValue: string | null = null; + /** Bumped whenever an attempt is started or invalidated; a settling attempt + * whose id no longer matches has been superseded/aborted and is ignored. */ + private activeAttemptId = 0; + + // --- readiness --- + private phaseValue: LifecyclePhase = "idle"; + /** True once a disconnect has happened since the last successful join, so the + * next success is reported as a recovery (`resynced`) not an initial `ready`. */ + private sawDisconnect = false; + private readonly lifecycleListeners = new Set(); + + constructor(deps: CoordinatorDeps) { + this.socket = deps.socket; + this.presenceId = deps.presenceId; + this.isConnected = deps.isConnected; + this.generateJoinId = deps.generateJoinId; + this.joinTimeoutMs = deps.joinTimeoutMs ?? DEFAULT_JOIN_TIMEOUT_MS; + this.debug = deps.debug ?? (() => undefined); + } + + // ---- observable state (replayable) ---- + + public get phase(): LifecyclePhase { + return this.phaseValue; + } + /** True only when the current room has been authoritatively confirmed - + * transport connectivity alone is never enough. */ + public get roomReady(): boolean { + return this.phaseValue === "ready"; + } + public get pendingJoinId(): string | null { + return this.pendingJoinIdValue; + } + public get currentRoom(): string | number | null { + return this.desiredRoom; + } + + public onLifecycle(listener: LifecycleListener): () => void { + this.lifecycleListeners.add(listener); + return () => this.lifecycleListeners.delete(listener); + } + + private emitLifecycle(event: LifecycleEvent): void { + this.lifecycleListeners.forEach(listener => listener(event)); + } + + // ---- intent API (called by SocketManager) ---- + + /** + * Records the intent to be in `room` and returns a promise that resolves + * when the room is authoritatively joined - possibly by a later attempt if + * the transport is down or drops mid-join. Supersedes any prior intent. + */ + public requestRoom(room: string | number, token: string): Promise { + // A newer intent supersedes the previous one: reject its still-pending + // caller so nobody waits forever on an abandoned room. + this.rejectLogical(new Error("JOIN cancelled: superseded")); + + this.intentGeneration += 1; + const generation = this.intentGeneration; + this.desiredRoom = room; + this.desiredToken = token; + this.phaseValue = "joining"; + + const promise = new Promise((resolve, reject) => { + this.logicalResolve = resolve; + this.logicalReject = reject; + }); + + // Invalidate any in-flight attempt from the prior intent. + this.abortCurrentAttempt("superseded"); + + if (this.isConnected()) { + this.startAttempt(generation); + } else { + // A3: do not emit a wire JOIN while disconnected - Socket.IO would buffer + // it and flush it on reconnect, possibly against a room we no longer want. + // The intent waits; `handleConnect` will start the attempt. + this.debug("requestRoom while disconnected - deferring JOIN until connect", room); + } + + return promise; + } + + /** + * Clears the current room intent so a later automatic reconnect cannot + * silently rejoin an abandoned room. Optionally scoped to a specific room so + * a stale teardown for room A cannot clear a newer intent for room B. + */ + public clearRoomIntent(expectedRoom?: string | number): void { + if ( + expectedRoom !== undefined && + this.desiredRoom !== null && + `${this.desiredRoom}` !== `${expectedRoom}` + ) { + // The intent has already moved on to a newer room - leave it alone. + return; + } + this.intentGeneration += 1; + this.desiredRoom = null; + this.desiredToken = null; + this.sawDisconnect = false; + this.abortCurrentAttempt("cleared"); + this.rejectLogical(new Error("JOIN cancelled: cleared")); + this.phaseValue = "idle"; + } + + // ---- transport events (called by SocketManager) ---- + + /** A transport connection was established (initial or a reconnect). */ + public handleConnect(): void { + if (this.desiredRoom == null || this.desiredToken == null) { + // No room wanted yet (e.g. initial connect before the page joins) - + // nothing to (re)join. + return; + } + if (this.phaseValue === "failed") { + // A prior attempt for this intent failed definitively (e.g. invalid + // token). Do NOT auto-retry on reconnect - that would be an unbounded + // retry loop against a request that cannot succeed. Recovery requires a + // fresh requestRoom (page re-navigation / reload), which resets the phase. + return; + } + this.phaseValue = "joining"; + this.startAttempt(this.intentGeneration); + } + + /** The transport dropped. Keep the logical intent pending for auto-rejoin. */ + public handleDisconnect(): void { + // Abort the in-flight attempt WITHOUT rejecting the logical intent - the + // reconnect will start a fresh attempt for the same desired room. + this.abortCurrentAttempt("disconnected"); + if (this.desiredRoom != null) { + this.sawDisconnect = true; + this.phaseValue = "disconnected"; + } else { + this.phaseValue = "idle"; + } + this.emitLifecycle("disconnected"); + } + + // ---- internals ---- + + private startAttempt(generation: number): void { + if (generation !== this.intentGeneration) return; // stale intent + if (this.desiredRoom == null || this.desiredToken == null) return; + + // Only one live attempt at a time. + this.abortCurrentAttempt("superseded"); + + const attemptId = (this.activeAttemptId += 1); + const joinId = this.generateJoinId(); + this.pendingJoinIdValue = joinId; + + const handle = joinRoomOverSocket( + this.socket, + this.desiredRoom, + this.desiredToken, + this.presenceId, + joinId, + this.joinTimeoutMs, + ); + this.currentAttempt = handle; + + handle.promise.then( + () => { + if (attemptId !== this.activeAttemptId) return; // superseded/aborted + this.onAttemptSuccess(generation, joinId); + }, + (err: Error) => { + if (attemptId !== this.activeAttemptId) return; // intentional cancel - ignore + this.onAttemptFailure(generation, joinId, err); + }, + ); + } + + /** + * Invalidates and cancels the in-flight attempt. Bumping `activeAttemptId` + * BEFORE cancelling means the cancellation's promise rejection is recognised + * as stale and ignored by the attached handler - so cancelling never looks + * like a real JOIN failure. + */ + private abortCurrentAttempt(reason: string): void { + if (!this.currentAttempt) { + this.pendingJoinIdValue = null; + return; + } + this.activeAttemptId += 1; + const handle = this.currentAttempt; + this.currentAttempt = null; + this.pendingJoinIdValue = null; + handle.cancel(reason); + } + + private onAttemptSuccess(generation: number, joinId: string): void { + if (generation !== this.intentGeneration) return; // superseded intent + if (this.pendingJoinIdValue === joinId) this.pendingJoinIdValue = null; + this.currentAttempt = null; + + const recovered = this.sawDisconnect; + this.sawDisconnect = false; + this.phaseValue = "ready"; + + // Resolve the original caller's logical promise (initial or via a + // reconnect-replacement attempt) exactly once. + const resolve = this.logicalResolve; + this.clearLogicalSettlers(); + if (resolve) resolve(); + + // `ready` on first join; `resynced` when we recovered from a disconnect, + // so only a genuine recovery surfaces a "reconnected" message. + this.emitLifecycle(recovered ? "resynced" : "ready"); + } + + private onAttemptFailure(generation: number, joinId: string, err: Error): void { + if (generation !== this.intentGeneration) return; // superseded intent + if (this.pendingJoinIdValue === joinId) this.pendingJoinIdValue = null; + this.currentAttempt = null; + + // A real failure: JOIN:error (e.g. invalid token) or a per-attempt timeout + // while connected. Reject the logical intent and stop - no auto-retry loop, + // no fake readiness. Recovery requires a fresh requestRoom (e.g. the page + // re-navigating or reloading). + this.phaseValue = "failed"; + this.rejectLogical(err); + this.emitLifecycle("failed"); + } + + private clearLogicalSettlers(): void { + this.logicalResolve = null; + this.logicalReject = null; + } + + private rejectLogical(err: Error): void { + const reject = this.logicalReject; + this.clearLogicalSettlers(); + if (reject) reject(err); + } +} diff --git a/spa/src/socket.ts b/spa/src/socket.ts index ec4e3ef1..7d39218d 100644 --- a/spa/src/socket.ts +++ b/spa/src/socket.ts @@ -1,7 +1,11 @@ import * as SocketIO from "socket.io-client"; import { debugMsg } from '@/helpers'; -import { joinRoomOverSocket } from "./join-protocol"; +import { + ReconnectCoordinator, + LifecycleEvent, + LifecyclePhase, +} from "./reconnect-coordinator"; /** * Generates a random per-tab presence id. Held only in memory for the @@ -13,8 +17,20 @@ function generatePresenceId(): string { return `${Date.now().toString(36)}-${Math.random().toString(36).slice(2)}`; } +/** Mints a unique id for a single JOIN attempt. */ +function generateJoinId(): string { + return `j-${Date.now().toString(36)}-${Math.random().toString(36).slice(2)}`; +} + +/** + * Thin adapter over the socket.io transport. It owns nothing about the join + * lifecycle itself - that lives in {@link ReconnectCoordinator} - it only + * creates the socket, forwards transport events into the coordinator, and + * exposes the coordinator's intent API and readiness state to the Vue layer. + */ class SocketManager { private socket: SocketIO.Socket; + private coordinator: ReconnectCoordinator; private readonly presenceIdValue: string = generatePresenceId(); constructor() {} @@ -22,14 +38,16 @@ class SocketManager { /** * The random per-tab presence id for this page instance. Combined with * the JWT-derived member id server-side to form the logical presence key - * `memberId:presenceId` - never the transport-level socket id. + * `memberId:presenceId` - never the transport-level socket id. Stable + * across reconnects for the tab's lifetime. */ public get presenceId(): string { return this.presenceIdValue; } /** - * Determines if the socket is currently connected. + * Determines if the socket transport is currently connected. Note this is + * NOT the same as room readiness - see {@link roomReady}. * @return `true` if a socket exists and it's connected, `false` otherwise */ public get connected(): boolean { @@ -37,11 +55,36 @@ class SocketManager { return this.socket.connected; } + /** + * Whether the current room has been authoritatively confirmed (a matching + * ROOM_STATE received). Transport connectivity alone is never enough - a + * reconnected-but-not-yet-resynced socket reports `false`. + */ + public get roomReady(): boolean { + return this.coordinator ? this.coordinator.roomReady : false; + } + + /** The current readiness phase, replayable by a late-subscribing consumer. */ + public get lifecyclePhase(): LifecyclePhase { + return this.coordinator ? this.coordinator.phase : "idle"; + } + + /** The joinId of the in-flight attempt, or null; used to correlate the + * persistent ROOM_STATE listener so a stale attempt can't re-reconcile. */ + public get pendingJoinId(): string | null { + return this.coordinator ? this.coordinator.pendingJoinId : null; + } + + /** The room the client currently intends to be in (used to room-tag AV). */ + public get currentRoom(): string | number | null { + return this.coordinator ? this.coordinator.currentRoom : null; + } + /** * Emits the given event on the socket, if it exists. * @param event name of event to emit * @param args 0-N items to send with the event - * @returns socket instace + * @returns socket instance */ public emit(event: string, ...args: any[]): SocketIO.Socket { if (!this.socket) return; @@ -49,24 +92,54 @@ class SocketManager { } /** - * Joins the room with the given room id and waits for the server's - * authoritative confirmation (a matching ROOM_STATE) before resolving. - * Rejects on JOIN:error or if no response arrives within the timeout - - * callers must not treat this as "joined" until it resolves. + * Emits a room-scoped AV (avatar movement/gesture/viewpoint) payload. Dropped + * entirely while disconnected so nothing is buffered by Socket.IO and flushed + * into a later room, and volatile so stale movement is never queued. The + * authoritative current room is stamped on so the server can reject any AV + * that doesn't match the socket's current room. + * @param payload the AV detail to broadcast + */ + public sendAv(payload: Record): void { + if (!this.socket || !this.socket.connected) return; // drop while disconnected + const room = this.coordinator ? this.coordinator.currentRoom : null; + if (room == null) return; + this.socket.volatile.emit("AV", { ...payload, room }); + } + + /** + * Records the intent to join the room and resolves once the server confirms + * with a matching ROOM_STATE. Unlike a raw transport emit, this intent + * survives a mid-join disconnect: it is retried automatically on reconnect + * and the returned promise resolves off whichever attempt succeeds. Rejects + * only on a definitive failure (invalid token / timeout) or if superseded by + * a newer room / cleared. * @param roomId id of room to join * @param userToken user's unique token - * @returns promise resolved once the server confirms the join, rejected on error/timeout + * @returns promise resolved once the server confirms the (possibly retried) join */ public joinRoom(roomId: string|number, userToken: string): Promise { - return joinRoomOverSocket(this.socket, roomId, userToken, this.presenceIdValue); + return this.coordinator.requestRoom(roomId, userToken); } /** - * Tells the server to unsubscribe the socket from the room with the given id. + * Clears room intent (so an automatic reconnect can't silently rejoin) and, + * only if currently connected, tells the server to unsubscribe. A disconnected + * transport has no live room membership to leave, so no `unsubscribe` is + * queued (which Socket.IO would otherwise flush on reconnect). * @param roomId id of room to leave */ public leaveRoom(roomId: string|number): void { - this.socket.emit("unsubscribe", { room: roomId }); + const wasConnected = this.connected; + this.coordinator.clearRoomIntent(roomId); + if (wasConnected) this.socket.emit("unsubscribe", { room: roomId }); + } + + /** + * Subscribes to room-readiness lifecycle transitions (disconnected / ready / + * resynced / failed). Returns an unsubscribe function. + */ + public onLifecycle(listener: (event: LifecycleEvent) => void): () => void { + return this.coordinator.onLifecycle(listener); } /** @@ -91,6 +164,14 @@ class SocketManager { return this.socket.off(event, callback); } + /** + * Creates the underlying socket. Extracted so tests can inject a fake + * transport without opening a real connection. + */ + protected createSocket(): SocketIO.Socket { + return SocketIO.io(); + } + /** * Creates and connects a socket instance. * @returns promise to be resolved on connection @@ -98,26 +179,34 @@ class SocketManager { public start(): Promise { if (this.socket) return; debugMsg("starting socket..."); - this.socket = SocketIO.io(); + this.socket = this.createSocket(); + this.coordinator = new ReconnectCoordinator({ + socket: this.socket, + presenceId: this.presenceIdValue, + isConnected: () => !!this.socket && this.socket.connected, + generateJoinId, + debug: debugMsg, + }); + // In socket.io v4 the reconnection-lifecycle events fire on the MANAGER + // (`socket.io`), not the socket instance; `connect` re-fires on the socket + // instance after every reconnect. So we drive (re)join from the instance + // `connect` and keep the manager `reconnect` only for debug visibility. this.socket.on("connect", () => this.onConnect()); this.socket.on("disconnect", () => this.onDisconnect()); - this.socket.on("reconnect", () => this.onReconnect()); - return new Promise(resolve => this.socket.on("connect", resolve)); + this.socket.io.on("reconnect", () => debugMsg("manager reconnect")); + return new Promise(resolve => this.socket.on("connect", () => resolve())); } - /** Connection event handler */ + /** Connection event handler - initial connect and every reconnect. */ private onConnect(): void { debugMsg('connect'); + this.coordinator.handleConnect(); } - /** Disconnection event handler */ + /** Disconnection event handler. */ private onDisconnect(): void { debugMsg('disconnected...'); - } - - /** Reconnection event handler */ - private onReconnect(): void { - debugMsg('reconnecting..'); + this.coordinator.handleDisconnect(); } } export { SocketManager }; From 94c1f6ae90a598f1517b1a579fa728fde9f519de Mon Sep 17 00:00:00 2001 From: Ryan Bundy Date: Tue, 21 Jul 2026 06:25:51 -0400 Subject: [PATCH 2/6] feat: rejoin and resync presence after reconnect Wire the world/chat UI to the reconnect lifecycle so a dropped socket recovers without a page refresh (issue #69). - WorldBrowserPage: correlate the persistent ROOM_STATE reconcile by joinId so a stale attempt can't re-reconcile; drop presence (AV/AV:new/AV:del) events whose room != the active room; route AV emits through the room-tagged, offline-safe sendAv; on a reconnect 'resynced' transition, re-announce the current viewpoint once so a restarted socket server relearns our stationary position. - Chat: drive input liveness from the room-readiness lifecycle instead of raw transport - input is disabled on disconnect and re-enabled only after a successful resync (a matching authoritative ROOM_STATE), never on mere reconnect. Gate initial enable on roomReady so a Chat mounted mid-resync stays disabled and learns the state on 'resynced'. One disconnected/reconnected message per outage/recovery; lifecycle subscription cleaned up on teardown. --- spa/src/components/Chat.vue | 48 +++++++++++++++---- .../pages/world-browser/WorldBrowserPage.vue | 45 +++++++++++++---- 2 files changed, 74 insertions(+), 19 deletions(-) diff --git a/spa/src/components/Chat.vue b/spa/src/components/Chat.vue index 42cab07e..e4710978 100644 --- a/spa/src/components/Chat.vue +++ b/spa/src/components/Chat.vue @@ -306,6 +306,7 @@ interface ChatData { pingIntervalId: any; worldMembers: any[]; chatEnabled: boolean; + unsubscribeLifecycle: (() => void) | null; showRole: boolean; showXP: boolean; tts: boolean; @@ -403,6 +404,7 @@ export default Vue.extend { + if (event === "resynced") this.sendInitialViewpoint(); + }); }, beforeDestroy() {}, async beforeCreate() { From 9208e7fc6027c851904a530a20516b1a12db2858 Mon Sep 17 00:00:00 2001 From: Ryan Bundy Date: Tue, 21 Jul 2026 06:26:03 -0400 Subject: [PATCH 3/6] test: cover socket reconnect lifecycle Extend the dependency-free harness (no new test framework) with three suites, run in deterministic order via tests/run-all.js (any failure exits non-zero): - presence.test.ts: correlated joinId echo/rejection, stale room+joinId responses ignored, and isPresenceEventForRoom room-guarding. - reconnect-coordinator.test.ts: drives the real coordinator via a fake socket + injectable isConnected/joinId - interrupted-initial-join recovery resolving the original caller, offline JOIN deferral, latest-room-wins (A interrupted then B offline), clear-while-offline prevents rejoin, invalid-auth no-retry-loop, stale ROOM_STATE can't flip readiness, connected-but-resyncing stays unready, one disconnected/resynced per cycle, flat lifecycle subscriptions. - server-presence.test.ts: boots the real server.js on an ephemeral port with signed JWTs (unconditional subprocess teardown) - joinId echo, malformed payload safety, duplicate-JOIN idempotency, room-move announce, new-socket rebind transform preservation, stale old-socket cannot delete/AV:del/AV a replacement presence, and AV room-tag rejection. All suites green (49 tests); npm run build clean under Node 14. --- spa/package.json | 2 +- spa/tests/presence.test.ts | 124 ++++++--- spa/tests/reconnect-coordinator.test.ts | 314 ++++++++++++++++++++++ spa/tests/run-all.js | 25 ++ spa/tests/server-presence.test.ts | 338 ++++++++++++++++++++++++ 5 files changed, 764 insertions(+), 39 deletions(-) create mode 100644 spa/tests/reconnect-coordinator.test.ts create mode 100644 spa/tests/run-all.js create mode 100644 spa/tests/server-presence.test.ts diff --git a/spa/package.json b/spa/package.json index e8b368ee..fc24ac15 100644 --- a/spa/package.json +++ b/spa/package.json @@ -8,7 +8,7 @@ "lint": "vue-cli-service lint", "dev": "vue-cli-service build --mode development --watch", "dev-server": "nodemon --inspect=0.0.0.0:9230 -r dotenv/config server.js", - "test": "tsc --project tests/tsconfig.json && node tests/.compiled/tests/presence.test.js" + "test": "tsc --project tests/tsconfig.json && node tests/run-all.js" }, "dependencies": { "axios": "^0.23.0", diff --git a/spa/tests/presence.test.ts b/spa/tests/presence.test.ts index 9f8482aa..a527e09c 100644 --- a/spa/tests/presence.test.ts +++ b/spa/tests/presence.test.ts @@ -8,7 +8,7 @@ */ import assert from "assert"; import { EventEmitter } from "events"; -import { PresenceStore, presenceKey, isSelfPresence, Presence } from "../src/presence"; +import { PresenceStore, presenceKey, isSelfPresence, isPresenceEventForRoom, Presence } from "../src/presence"; import { joinRoomOverSocket } from "../src/join-protocol"; type Test = { name: string; run: () => void | Promise }; @@ -191,66 +191,114 @@ test("isSelfPresence does not match a different member using the same tab id", ( assert.strictEqual(isSelfPresence(makePresence({ memberId: 1, presenceId: "tab-a" }), 2, "tab-a"), false); }); -describe("joinRoomOverSocket", () => { - function makeFakeSocket() { - const emitter = new EventEmitter(); - const emitted: any[] = []; - const socket = { - on: (event: string, cb: (...args: any[]) => void) => emitter.on(event, cb), - off: (event: string, cb: (...args: any[]) => void) => emitter.off(event, cb), - emit: (event: string, ...args: any[]) => { - emitted.push({ event, args }); - if (event === "JOIN") return; // client->server emit, not looped back automatically - emitter.emit(event, ...args); - }, - }; - return { socket, emitter, emitted }; - } +function makeFakeSocket() { + const emitter = new EventEmitter(); + const emitted: any[] = []; + const socket = { + on: (event: string, cb: (...args: any[]) => void) => emitter.on(event, cb), + off: (event: string, cb: (...args: any[]) => void) => emitter.off(event, cb), + emit: (event: string, ...args: any[]) => { + emitted.push({ event, args }); + if (event === "JOIN") return; // client->server emit, not looped back automatically + emitter.emit(event, ...args); + }, + }; + const lastJoin = () => emitted.filter(e => e.event === "JOIN").slice(-1)[0]?.args[0]; + return { socket, emitter, emitted, lastJoin }; +} - test("resolves once the server confirms with a ROOM_STATE for the requested room", async () => { - const { socket, emitter, emitted } = makeFakeSocket(); - const promise = joinRoomOverSocket(socket, "room-1", "token", "presence-1"); +describe("joinRoomOverSocket (correlated JOIN)", () => { + test("emits JOIN with the joinId and resolves on a matching room+joinId ROOM_STATE", async () => { + const { socket, emitter, lastJoin } = makeFakeSocket(); + const handle = joinRoomOverSocket(socket, "room-1", "token", "presence-1", "join-1"); + + assert.deepStrictEqual(lastJoin(), { + room: "room-1", token: "token", presenceId: "presence-1", joinId: "join-1", + }); + + emitter.emit("ROOM_STATE", { room: "room-1", joinId: "join-1", presences: [] }); + await handle.promise; + }); - assert.strictEqual(emitted[0].event, "JOIN"); - assert.deepStrictEqual(emitted[0].args[0], { room: "room-1", token: "token", presenceId: "presence-1" }); + test("a ROOM_STATE with the right room but a stale joinId does not resolve", async () => { + const { socket, emitter } = makeFakeSocket(); + const handle = joinRoomOverSocket(socket, "room-1", "token", "presence-1", "join-2", 30); - emitter.emit("ROOM_STATE", { room: "room-1", presences: [] }); - await promise; + emitter.emit("ROOM_STATE", { room: "room-1", joinId: "join-1", presences: [] }); // older attempt + await assert.rejects(handle.promise, /timed out/); // never confirmed -> times out }); test("ignores a ROOM_STATE for a different room and keeps waiting", async () => { const { socket, emitter } = makeFakeSocket(); - const promise = joinRoomOverSocket(socket, "room-1", "token", "presence-1"); + const handle = joinRoomOverSocket(socket, "room-1", "token", "presence-1", "join-1"); - emitter.emit("ROOM_STATE", { room: "some-other-room", presences: [] }); - emitter.emit("ROOM_STATE", { room: "room-1", presences: [] }); + emitter.emit("ROOM_STATE", { room: "some-other-room", joinId: "join-1", presences: [] }); + emitter.emit("ROOM_STATE", { room: "room-1", joinId: "join-1", presences: [] }); - await promise; // must not hang/reject - the second, matching event resolves it + await handle.promise; // second, matching event resolves it }); - test("rejects on JOIN:error instead of resolving", async () => { + test("rejects on a matching room+joinId JOIN:error", async () => { const { socket, emitter } = makeFakeSocket(); - const promise = joinRoomOverSocket(socket, "room-1", "token", "presence-1"); + const handle = joinRoomOverSocket(socket, "room-1", "token", "presence-1", "join-1"); + + emitter.emit("JOIN:error", { room: "room-1", joinId: "join-1", reason: "invalid_token" }); + + await assert.rejects(handle.promise, /invalid_token/); + }); - emitter.emit("JOIN:error", { reason: "invalid_token" }); + test("ignores a JOIN:error correlated to a different (older) attempt", async () => { + const { socket, emitter } = makeFakeSocket(); + const handle = joinRoomOverSocket(socket, "room-1", "token", "presence-1", "join-2", 30); - await assert.rejects(promise, /invalid_token/); + emitter.emit("JOIN:error", { room: "room-1", joinId: "join-1", reason: "invalid_token" }); + await assert.rejects(handle.promise, /timed out/); // stale error ignored -> times out }); test("rejects if no response arrives before the timeout", async () => { const { socket } = makeFakeSocket(); - const promise = joinRoomOverSocket(socket, "room-1", "token", "presence-1", 20); + const handle = joinRoomOverSocket(socket, "room-1", "token", "presence-1", "join-1", 20); + + await assert.rejects(handle.promise, /timed out/); + }); + + test("cancel() rejects the attempt with its reason and stops listening", async () => { + const { socket, emitter } = makeFakeSocket(); + const handle = joinRoomOverSocket(socket, "room-1", "token", "presence-1", "join-1"); - await assert.rejects(promise, /timed out/); + handle.cancel("superseded"); + await assert.rejects(handle.promise, /cancelled: superseded/); + // A later matching ROOM_STATE must not throw or double-settle. + emitter.emit("ROOM_STATE", { room: "room-1", joinId: "join-1", presences: [] }); }); - test("a late ROOM_STATE after timeout/rejection does not resolve the already-settled promise", async () => { + test("a late ROOM_STATE after a settled attempt does not re-settle it", async () => { const { socket, emitter } = makeFakeSocket(); - const promise = joinRoomOverSocket(socket, "room-1", "token", "presence-1", 20); + const handle = joinRoomOverSocket(socket, "room-1", "token", "presence-1", "join-1", 20); + + await assert.rejects(handle.promise); + emitter.emit("ROOM_STATE", { room: "room-1", joinId: "join-1", presences: [] }); + }); +}); + +describe("isPresenceEventForRoom", () => { + test("accepts an event tagged for the active room (normalized compare)", () => { + assert.strictEqual(isPresenceEventForRoom("room-1", "room-1"), true); + assert.strictEqual(isPresenceEventForRoom(5, "5"), true); + }); + + test("rejects an event for a non-active room", () => { + assert.strictEqual(isPresenceEventForRoom("room-2", "room-1"), false); + }); + + test("rejects an untagged (missing-room) event rather than accepting it ambiguously", () => { + assert.strictEqual(isPresenceEventForRoom(undefined, "room-1"), false); + assert.strictEqual(isPresenceEventForRoom(null, "room-1"), false); + }); - await assert.rejects(promise); - // Should not throw or double-resolve - listeners were cleaned up on timeout. - emitter.emit("ROOM_STATE", { room: "room-1", presences: [] }); + test("rejects any event when there is no active room", () => { + assert.strictEqual(isPresenceEventForRoom("room-1", undefined), false); + assert.strictEqual(isPresenceEventForRoom("room-1", null), false); }); }); diff --git a/spa/tests/reconnect-coordinator.test.ts b/spa/tests/reconnect-coordinator.test.ts new file mode 100644 index 00000000..b1660368 --- /dev/null +++ b/spa/tests/reconnect-coordinator.test.ts @@ -0,0 +1,314 @@ +/** + * Dependency-free tests for the ReconnectCoordinator - the owner of the + * room-join lifecycle. Same hand-rolled harness as presence.test.ts (no test + * framework: this toolchain destabilizes if jest is added). The coordinator is + * driven through a fake EmitterLike socket plus an injectable `isConnected` + * probe and deterministic joinId generator, so these exercise the REAL state + * machine (not a test-only imitation) - which is why the machine was extracted + * into its own `@/`-free module the Node harness can import directly. + */ +import assert from "assert"; +import { EventEmitter } from "events"; +import { ReconnectCoordinator, LifecycleEvent } from "../src/reconnect-coordinator"; + +type Test = { name: string; run: () => void | Promise }; +const tests: Test[] = []; +function test(name: string, run: () => void | Promise): void { + tests.push({ name, run }); +} + +/** Flush pending microtasks (the coordinator settles attempts on a microtask). */ +const tick = (): Promise => new Promise(resolve => setImmediate(resolve)); + +/** Tracks a promise's settlement without leaving an unhandled rejection. */ +function track(p: Promise) { + const state: { done: boolean; error: Error | null } = { done: false, error: null }; + p.then( + () => { state.done = true; }, + (err: Error) => { state.done = true; state.error = err; }, + ); + return state; +} + +function setup(opts: { connected?: boolean; timeout?: number } = {}) { + const emitter = new EventEmitter(); + const emitted: { event: string; args: any[] }[] = []; + let connected = opts.connected ?? true; + let joinCounter = 0; + + const socket = { + on: (event: string, cb: (...args: any[]) => void) => emitter.on(event, cb), + off: (event: string, cb: (...args: any[]) => void) => emitter.off(event, cb), + emit: (event: string, ...args: any[]) => { + emitted.push({ event, args }); + // JOIN is client->server only; responses are injected via the emitter. + }, + }; + + const events: LifecycleEvent[] = []; + const coord = new ReconnectCoordinator({ + socket, + presenceId: "pres-1", + isConnected: () => connected, + generateJoinId: () => `jid-${++joinCounter}`, + joinTimeoutMs: opts.timeout ?? 10000, + }); + coord.onLifecycle(e => events.push(e)); + + const joins = () => emitted.filter(e => e.event === "JOIN").map(e => e.args[0]); + const lastJoin = () => joins().slice(-1)[0]; + const setConnected = (v: boolean) => { connected = v; }; + const roomState = (room: string | number, joinId: string, presences: any[] = []) => + emitter.emit("ROOM_STATE", { room, joinId, presences }); + const joinError = (room: string | number, joinId: string, reason: string) => + emitter.emit("JOIN:error", { room, joinId, reason }); + + return { coord, emitter, emitted, events, joins, lastJoin, setConnected, roomState, joinError }; +} + +test("initial join emits a correlated JOIN, resolves the caller, and reports ready", async () => { + const s = setup(); + const p = track(s.coord.requestRoom("room-A", "token")); + assert.deepStrictEqual(s.lastJoin(), { + room: "room-A", token: "token", presenceId: "pres-1", joinId: "jid-1", + }); + assert.strictEqual(s.coord.roomReady, false); // not ready until confirmed + + s.roomState("room-A", "jid-1"); + await tick(); + + assert.strictEqual(p.done && !p.error, true); + assert.strictEqual(s.coord.roomReady, true); + assert.strictEqual(s.coord.phase, "ready"); + assert.deepStrictEqual(s.events, ["ready"]); // "ready", NOT "resynced", on first join +}); + +test("requestRoom while disconnected defers the wire JOIN until connect (no offline buffering)", async () => { + const s = setup({ connected: false }); + const p = track(s.coord.requestRoom("room-A", "token")); + + assert.strictEqual(s.joins().length, 0); // A3: nothing emitted while disconnected + + s.setConnected(true); + s.coord.handleConnect(); + assert.strictEqual(s.joins().length, 1); + assert.strictEqual(s.lastJoin().room, "room-A"); + + s.roomState("room-A", s.lastJoin().joinId); + await tick(); + assert.strictEqual(p.done && !p.error, true); + assert.strictEqual(s.coord.roomReady, true); +}); + +test("an interrupted initial JOIN recovers on reconnect and resolves the ORIGINAL caller", async () => { + const s = setup(); + const p = track(s.coord.requestRoom("room-A", "token")); + const firstJoinId = s.lastJoin().joinId; + + // Transport drops before ROOM_STATE arrives. + s.coord.handleDisconnect(); + await tick(); + assert.strictEqual(p.done, false); // logical intent stays pending + assert.strictEqual(s.coord.phase, "disconnected"); + assert.deepStrictEqual(s.events, ["disconnected"]); + + // Reconnect -> a fresh attempt for the SAME room, same presenceId, new joinId. + s.coord.handleConnect(); + const secondJoin = s.lastJoin(); + assert.strictEqual(secondJoin.room, "room-A"); + assert.strictEqual(secondJoin.presenceId, "pres-1"); + assert.notStrictEqual(secondJoin.joinId, firstJoinId); + assert.strictEqual(s.coord.roomReady, false); // connected but not yet resynced (A7) + + s.roomState("room-A", secondJoin.joinId); + await tick(); + assert.strictEqual(p.done && !p.error, true); // original caller finally resolves + assert.strictEqual(s.coord.roomReady, true); + assert.deepStrictEqual(s.events, ["disconnected", "resynced"]); // recovery, not a 2nd "ready" +}); + +test("a newer room intent supersedes the older pending one", async () => { + const s = setup(); + const pA = track(s.coord.requestRoom("room-A", "token")); + const pB = track(s.coord.requestRoom("room-B", "token")); + await tick(); + + assert.strictEqual(!!pA.error, true); + assert.ok(/superseded/.test(pA.error!.message)); + assert.strictEqual(s.lastJoin().room, "room-B"); + assert.strictEqual(s.coord.pendingJoinId, s.lastJoin().joinId); + + s.roomState("room-B", s.lastJoin().joinId); + await tick(); + assert.strictEqual(pB.done && !pB.error, true); +}); + +test("A interrupted then B selected while offline: only B is joined on reconnect, A is superseded", async () => { + const s = setup(); + const pA = track(s.coord.requestRoom("room-A", "token")); + assert.strictEqual(s.joins().length, 1); // A's initial attempt + + // Transport drops, then the user navigates to B while still offline. + s.setConnected(false); + s.coord.handleDisconnect(); + const pB = track(s.coord.requestRoom("room-B", "token")); + await tick(); + + assert.strictEqual(!!pA.error, true); // A rejected as superseded + assert.ok(/superseded/.test(pA.error!.message)); + assert.strictEqual(s.joins().length, 1); // still nothing new emitted while offline + + // Reconnect: exactly one new attempt, and it is for B (never a stale A rejoin). + s.setConnected(true); + s.coord.handleConnect(); + assert.strictEqual(s.joins().length, 2); + assert.strictEqual(s.lastJoin().room, "room-B"); + assert.strictEqual(s.joins().filter(j => j.room === "room-A").length, 1); // only the pre-drop A + + s.roomState("room-B", s.lastJoin().joinId); + await tick(); + assert.strictEqual(pB.done && !pB.error, true); +}); + +test("clearing room intent while offline prevents any later automatic rejoin", async () => { + const s = setup(); + const pA = track(s.coord.requestRoom("room-A", "token")); + s.setConnected(false); + s.coord.handleDisconnect(); + + s.coord.clearRoomIntent(); + await tick(); + assert.strictEqual(!!pA.error, true); + assert.ok(/cleared/.test(pA.error!.message)); + + const joinsBefore = s.joins().length; + s.setConnected(true); + s.coord.handleConnect(); // must NOT rejoin the abandoned room + assert.strictEqual(s.joins().length, joinsBefore); + assert.strictEqual(s.coord.phase, "idle"); + assert.strictEqual(s.coord.currentRoom, null); +}); + +test("clearRoomIntent scoped to a room does not clear a newer intent for a different room", () => { + const s = setup(); + track(s.coord.requestRoom("room-A", "token")); // rejected (superseded) - tracked + track(s.coord.requestRoom("room-B", "token")); // intent moved to B + s.coord.clearRoomIntent("room-A"); // stale teardown for A - must be a no-op now + assert.strictEqual(s.coord.currentRoom, "room-B"); // B intent preserved + s.coord.clearRoomIntent("room-B"); + assert.strictEqual(s.coord.currentRoom, null); +}); + +test("invalid auth fails the join and does NOT trigger an auto-retry loop on reconnect", async () => { + const s = setup(); + const p = track(s.coord.requestRoom("room-A", "token")); + const joinId = s.lastJoin().joinId; + + s.joinError("room-A", joinId, "invalid_token"); + await tick(); + assert.strictEqual(!!p.error, true); + assert.ok(/invalid_token/.test(p.error!.message)); + assert.strictEqual(s.coord.phase, "failed"); + assert.deepStrictEqual(s.events, ["failed"]); + + // A subsequent reconnect must not silently re-attempt the doomed request. + s.coord.handleConnect(); + assert.strictEqual(s.joins().length, 1); // still just the one attempt + assert.strictEqual(s.events.includes("resynced"), false); +}); + +test("a stale ROOM_STATE from a superseded attempt cannot flip readiness", async () => { + const s = setup(); + const pA = track(s.coord.requestRoom("room-A", "token")); + const staleJoinId = s.lastJoin().joinId; + const pB = track(s.coord.requestRoom("room-B", "token")); + const freshJoinId = s.lastJoin().joinId; + await tick(); + assert.ok(pA.error); // A superseded + + // A late ROOM_STATE for the OLD attempt must not resolve B. + s.roomState("room-A", staleJoinId); + await tick(); + assert.strictEqual(s.coord.roomReady, false); + assert.strictEqual(pB.done, false); + + s.roomState("room-B", freshJoinId); + await tick(); + assert.strictEqual(pB.done && !pB.error, true); + assert.strictEqual(s.coord.roomReady, true); +}); + +test("roomReady stays false while connected but still resyncing (transport alone is not enough)", async () => { + const s = setup(); + const p = track(s.coord.requestRoom("room-A", "token")); + s.roomState("room-A", s.lastJoin().joinId); + await tick(); + assert.strictEqual(s.coord.roomReady, true); + + s.coord.handleDisconnect(); + assert.strictEqual(s.coord.roomReady, false); + s.coord.handleConnect(); // transport back, but not yet confirmed + assert.strictEqual(s.coord.roomReady, false); + assert.strictEqual(s.coord.phase, "joining"); + + s.roomState("room-A", s.lastJoin().joinId); + await tick(); + assert.strictEqual(s.coord.roomReady, true); + void p; +}); + +test("a full outage/recovery cycle emits exactly one disconnected and one resynced", async () => { + const s = setup(); + const p = track(s.coord.requestRoom("room-A", "token")); + s.roomState("room-A", s.lastJoin().joinId); + await tick(); + + s.coord.handleDisconnect(); + s.coord.handleConnect(); + s.roomState("room-A", s.lastJoin().joinId); + await tick(); + + assert.deepStrictEqual(s.events, ["ready", "disconnected", "resynced"]); + void p; +}); + +test("lifecycle unsubscribe stops delivery and subscriptions do not accumulate", async () => { + const s = setup(); + let a = 0; + let b = 0; + const unsubA = s.coord.onLifecycle(() => { a += 1; }); + s.coord.onLifecycle(() => { b += 1; }); + + const p = track(s.coord.requestRoom("room-A", "token")); + s.roomState("room-A", s.lastJoin().joinId); + await tick(); + assert.strictEqual(a, 1); + assert.strictEqual(b, 1); + + unsubA(); + s.coord.handleDisconnect(); + assert.strictEqual(a, 1); // no further delivery after unsubscribe + assert.strictEqual(b, 2); + void p; +}); + +async function run(): Promise { + let failures = 0; + for (const { name, run: runTest } of tests) { + try { + await runTest(); + console.log(` ✓ ${name}`); + } catch (err) { + failures += 1; + console.error(` ✗ ${name}`); + console.error(err instanceof Error ? ` ${err.message}` : err); + } + } + + console.log(`\n${tests.length - failures}/${tests.length} passed`); + if (failures > 0) { + process.exit(1); + } +} + +run(); diff --git a/spa/tests/run-all.js b/spa/tests/run-all.js new file mode 100644 index 00000000..24b77cbe --- /dev/null +++ b/spa/tests/run-all.js @@ -0,0 +1,25 @@ +/** + * Runs every compiled test suite in a deterministic order, in its own Node + * process (so one suite's `process.exit` can't cut another short), and exits + * non-zero if ANY suite fails. Kept dependency-free - plain Node, no runner. + */ +const { spawnSync } = require("child_process"); +const path = require("path"); + +const SUITES = [ + "tests/.compiled/tests/presence.test.js", + "tests/.compiled/tests/reconnect-coordinator.test.js", + "tests/.compiled/tests/server-presence.test.js", +]; + +let failed = false; +for (const suite of SUITES) { + console.log(`\n=== ${suite} ===`); + const result = spawnSync(process.execPath, [path.resolve(suite)], { stdio: "inherit" }); + if (result.status !== 0 || result.error) { + failed = true; + if (result.error) console.error(` could not run ${suite}: ${result.error.message}`); + } +} + +process.exit(failed ? 1 : 0); diff --git a/spa/tests/server-presence.test.ts b/spa/tests/server-presence.test.ts new file mode 100644 index 00000000..b188d7b4 --- /dev/null +++ b/spa/tests/server-presence.test.ts @@ -0,0 +1,338 @@ +/** + * Real-protocol tests for server.js. These boot the ACTUAL socket server as a + * subprocess on an ephemeral port (never the dev stack's ports) and drive it + * with real socket.io-client connections and signed JWTs - so the server-side + * ownership guards, room-tagging, correlation and transform handling are proven + * over the wire, not against a re-implementation. socket.io / socket.io-client / + * jsonwebtoken are already dependencies, so no new dependency is introduced. + * + * Presence ownership is asserted only through protocol-visible snapshots and + * events (ROOM_STATE / AV:new / AV:del) - there is no test-only endpoint. + */ +import assert from "assert"; + +// Required via `require` (not `import`) so the suite compiles without extra +// @types packages; these are runtime-only helpers. +const { spawn } = require("child_process"); +const net = require("net"); +const path = require("path"); +const jwt = require("jsonwebtoken"); +const { io } = require("socket.io-client"); + +const SPA_DIR = path.resolve(__dirname, "../../.."); +const SERVER = path.join(SPA_DIR, "server.js"); +const SECRET = "test-secret-do-not-use-in-prod"; + +type Test = { name: string; run: () => Promise }; +const tests: Test[] = []; +function test(name: string, run: () => Promise): void { + tests.push({ name, run }); +} + +let PORT = 0; +let serverProc: any = null; +let serverLog = ""; + +function signToken(id: number, username: string): string { + return jwt.sign({ id, username, avatar: { id: `av-${id}` } }, SECRET); +} + +function getFreePort(): Promise { + return new Promise((resolve, reject) => { + const srv = net.createServer(); + srv.on("error", reject); + // Discover a free port via the OS (listen on 0), then hand the concrete + // number to the server - the server itself is never given the string "0". + srv.listen(0, "127.0.0.1", () => { + const port = srv.address().port; + srv.close(() => resolve(port)); + }); + }); +} + +function startServer(port: number): Promise { + return new Promise((resolve, reject) => { + const child = spawn(process.execPath, [SERVER], { + cwd: SPA_DIR, + env: { ...process.env, WEBSOCKET_PORT: String(port), JWT_SECRET: SECRET }, + stdio: ["ignore", "pipe", "pipe"], + }); + let settled = false; + const onOut = (d: Buffer) => { + serverLog += d.toString(); + if (!settled && serverLog.includes(`listening on port:${port}`)) { + settled = true; + clearTimeout(timer); + resolve(child); + } + }; + child.stdout.on("data", onOut); + child.stderr.on("data", (d: Buffer) => { serverLog += d.toString(); }); + child.on("exit", (code: number) => { + if (!settled) { + settled = true; + clearTimeout(timer); + reject(new Error(`server exited before ready (code ${code})`)); + } + }); + const timer = setTimeout(() => { + if (!settled) { + settled = true; + reject(new Error("timed out waiting for server ready")); + } + }, 10000); + }); +} + +function killServer(child: any): Promise { + return new Promise((resolve) => { + if (!child || child.killed) return resolve(); + let done = false; + const finish = () => { if (!done) { done = true; resolve(); } }; + child.on("exit", finish); + child.kill("SIGKILL"); + setTimeout(finish, 2000); + }); +} + +function connect(): Promise { + const sock = io(`http://127.0.0.1:${PORT}`, { + transports: ["websocket"], + reconnection: false, + forceNew: true, + }); + return new Promise((resolve, reject) => { + const t = setTimeout(() => reject(new Error("client connect timeout")), 5000); + sock.on("connect", () => { clearTimeout(t); resolve(sock); }); + sock.on("connect_error", (e: Error) => { clearTimeout(t); reject(e); }); + }); +} + +/** Waits for the next `event` matching an optional predicate. */ +function waitFor(sock: any, event: string, predicate?: (p: any) => boolean, timeoutMs = 3000): Promise { + return new Promise((resolve, reject) => { + const handler = (payload: any) => { + if (predicate && !predicate(payload)) return; + cleanup(); + resolve(payload); + }; + const timer = setTimeout(() => { cleanup(); reject(new Error(`timeout waiting for ${event}`)); }, timeoutMs); + function cleanup() { clearTimeout(timer); sock.off(event, handler); } + sock.on(event, handler); + }); +} + +/** Asserts `event` does NOT fire within the window (negative control). */ +function expectNone(sock: any, event: string, windowMs = 600): Promise { + return new Promise((resolve, reject) => { + const handler = (payload: any) => { cleanup(); reject(new Error(`unexpected ${event}: ${JSON.stringify(payload)}`)); }; + const timer = setTimeout(() => { cleanup(); resolve(); }, windowMs); + function cleanup() { clearTimeout(timer); sock.off(event, handler); } + sock.on(event, handler); + }); +} + +/** Emits a JOIN and resolves with the correlated ROOM_STATE. */ +async function join(sock: any, room: string, token: string, presenceId: string, joinId: string): Promise { + const rs = waitFor(sock, "ROOM_STATE", (p) => p.joinId === joinId); + sock.emit("JOIN", { room, token, presenceId, joinId }); + return rs; +} + +const sockets: any[] = []; +async function newClient(): Promise { + const sock = await connect(); + sockets.push(sock); + return sock; +} + +test("ROOM_STATE echoes the room and joinId of the attempt", async () => { + const token = signToken(1, "alice"); + const sock = await newClient(); + const rs = await join(sock, "room-echo", token, "pres-1", "jid-echo"); + assert.strictEqual(rs.room, "room-echo"); + assert.strictEqual(rs.joinId, "jid-echo"); + assert.ok(Array.isArray(rs.presences)); +}); + +test("an invalid token yields a JOIN:error echoing room and joinId", async () => { + const sock = await newClient(); + const err = waitFor(sock, "JOIN:error", (p) => p.joinId === "jid-bad"); + sock.emit("JOIN", { room: "room-x", token: "not-a-jwt", presenceId: "pres-1", joinId: "jid-bad" }); + const payload = await err; + assert.strictEqual(payload.room, "room-x"); + assert.strictEqual(payload.joinId, "jid-bad"); + assert.strictEqual(payload.reason, "invalid_token"); +}); + +test("a malformed JOIN payload is rejected without crashing the server", async () => { + const sock = await newClient(); + const err = waitFor(sock, "JOIN:error", (p) => p && p.reason === "invalid_payload"); + sock.emit("JOIN", null); + const payload = await err; + assert.strictEqual(payload.reason, "invalid_payload"); + // The server is still alive: a subsequent valid JOIN succeeds. + const token = signToken(2, "bob"); + const rs = await join(sock, "room-alive", token, "pres-alive", "jid-alive"); + assert.strictEqual(rs.room, "room-alive"); +}); + +test("a duplicate same-room JOIN does not re-announce the presence to peers", async () => { + const room = "room-dup"; + const peer = await newClient(); + await join(peer, room, signToken(10, "peer"), "pres-peer", "jid-p"); + + const subject = await newClient(); + const sawJoin = waitFor(peer, "AV:new", (p) => p.presenceId === "pres-sub"); + await join(subject, room, signToken(11, "sub"), "pres-sub", "jid-s1"); + const avNew = await sawJoin; + assert.strictEqual(avNew.room, room); // AV:new is room-tagged + + // Re-JOIN the same room/presence: peers must NOT get a second "someone joined". + const noSecond = expectNone(peer, "AV:new"); + await join(subject, room, signToken(11, "sub"), "pres-sub", "jid-s2"); + await noSecond; +}); + +test("a different-room JOIN announces departure to the old room and arrival to the new", async () => { + const roomA = "room-A-move"; + const roomB = "room-B-move"; + const peerA = await newClient(); + const peerB = await newClient(); + await join(peerA, roomA, signToken(20, "pa"), "pres-pa", "jid-pa"); + await join(peerB, roomB, signToken(21, "pb"), "pres-pb", "jid-pb"); + + const subject = await newClient(); + await join(subject, roomA, signToken(22, "mover"), "pres-mv", "jid-m1"); + + const leftA = waitFor(peerA, "AV:del", (p) => p.presenceId === "pres-mv"); + const enteredB = waitFor(peerB, "AV:new", (p) => p.presenceId === "pres-mv"); + await join(subject, roomB, signToken(22, "mover"), "pres-mv", "jid-m2"); + + const del = await leftA; + const add = await enteredB; + assert.strictEqual(del.room, roomA); // AV:del carries the OLD room + assert.strictEqual(add.room, roomB); // AV:new carries the NEW room +}); + +test("a rebind by a new socket preserves the transform and leaves one snapshot entry", async () => { + const room = "room-rebind"; + const sock1 = await newClient(); + await join(sock1, room, signToken(30, "carol"), "pres-rb", "jid-r1"); + // Move to a known position (room-tagged AV, from the owning socket). + sock1.emit("AV", { room, pos: [5, 6, 7], rot: [0, 1, 0, 2] }); + await new Promise((r) => setTimeout(r, 200)); // let the server store it + + // A NEW socket rebinds the same logical presence while it still exists. + const sock2 = await newClient(); + const rs = await join(sock2, room, signToken(30, "carol"), "pres-rb", "jid-r2"); + const mine = rs.presences.filter((p: any) => p.presenceId === "pres-rb"); + assert.strictEqual(mine.length, 1); // exactly one record, not a duplicate + assert.deepStrictEqual(mine[0].pos, [5, 6, 7]); // transform preserved across rebind + assert.deepStrictEqual(mine[0].rot, [0, 1, 0, 2]); +}); + +test("a stale old socket cannot delete or announce a presence a newer socket now owns", async () => { + const room = "room-stale"; + const peer = await newClient(); + await join(peer, room, signToken(40, "peer40"), "pres-peer40", "jid-pe"); + + const sock1 = await newClient(); + await join(sock1, room, signToken(41, "dana"), "pres-stale", "jid-o1"); + + // New socket takes over the same logical presence (a reconnect rebind). + const sock2 = await newClient(); + await join(sock2, room, signToken(41, "dana"), "pres-stale", "jid-o2"); + + // The stale old socket disconnecting must NOT broadcast an AV:del for the + // presence the new socket now owns. + const noDel = expectNone(peer, "AV:del", 800); + sock1.disconnect(); + await noDel; + + // The presence is still there: a fresh observer sees it in the snapshot. + const observer = await newClient(); + const rs = await join(observer, room, signToken(42, "obs"), "pres-obs", "jid-ob"); + const stillThere = rs.presences.some((p: any) => p.presenceId === "pres-stale"); + assert.strictEqual(stillThere, true); +}); + +test("a stale old socket cannot broadcast AV under a presence a newer socket owns", async () => { + const room = "room-staleav"; + const peer = await newClient(); + await join(peer, room, signToken(50, "peer50"), "pres-peer50", "jid-pv"); + + const sock1 = await newClient(); + await join(sock1, room, signToken(51, "eve"), "pres-av", "jid-a1"); + const sock2 = await newClient(); + await join(sock2, room, signToken(51, "eve"), "pres-av", "jid-a2"); + + // Stale socket's movement is ignored (it no longer owns the presence)... + const noAv = expectNone(peer, "AV", 700); + sock1.emit("AV", { room, pos: [9, 9, 9] }); + await noAv; + + // ...but the current owner's movement is relayed (positive control). + const gotAv = waitFor(peer, "AV", (p) => p.presenceId === "pres-av"); + sock2.emit("AV", { room, pos: [1, 2, 3] }); + const av = await gotAv; + assert.strictEqual(av.room, room); // relayed AV is room-tagged + assert.deepStrictEqual(av.pos, [1, 2, 3]); +}); + +test("an AV tagged for a different room than the socket's current room is dropped", async () => { + const room = "room-avguard"; + const peer = await newClient(); + await join(peer, room, signToken(60, "peer60"), "pres-peer60", "jid-g1"); + const subject = await newClient(); + await join(subject, room, signToken(61, "frank"), "pres-fr", "jid-g2"); + + // Mis-tagged AV (claims a room the socket isn't in) must be dropped. + const noAv = expectNone(peer, "AV", 700); + subject.emit("AV", { room: "some-other-room", pos: [7, 7, 7] }); + await noAv; + + // Correctly-tagged AV is relayed (positive control). + const gotAv = waitFor(peer, "AV", (p) => p.presenceId === "pres-fr"); + subject.emit("AV", { room, pos: [4, 5, 6] }); + const av = await gotAv; + assert.deepStrictEqual(av.pos, [4, 5, 6]); +}); + +async function run(): Promise { + let failures = 0; + try { + PORT = await getFreePort(); + serverProc = await startServer(PORT); + } catch (err) { + console.error(" ✗ could not start test server"); + console.error(err instanceof Error ? ` ${err.message}` : err); + if (serverLog) console.error(" --- server output ---\n" + serverLog); + process.exit(1); + } + + for (const { name, run: runTest } of tests) { + try { + await runTest(); + console.log(` ✓ ${name}`); + } catch (err) { + failures += 1; + console.error(` ✗ ${name}`); + console.error(err instanceof Error ? ` ${err.message}` : err); + } + } + + // Unconditional teardown: close every client, then kill the server subprocess. + for (const s of sockets) { + try { s.disconnect(); } catch (e) { /* ignore */ } + } + await killServer(serverProc); + + console.log(`\n${tests.length - failures}/${tests.length} passed`); + if (failures > 0) { + if (serverLog) console.error("--- server output ---\n" + serverLog); + process.exit(1); + } +} + +run(); From 3275e1545356074d31548f9fa5ef40b860daa2c4 Mon Sep 17 00:00:00 2001 From: Ryan Bundy Date: Tue, 21 Jul 2026 06:42:31 -0400 Subject: [PATCH 4/6] fix: address reconnect review findings Local Copilot review adjudication - fixes for the three reproduced/cheap items; two findings were adjudicated as no-change (documented in the PR report). - server.js: an invalid joinId JOIN:error now echoes the client's raw joinId instead of undefined, so the client correlates and fails the attempt fast rather than waiting out the timeout. - socket.ts: leaveRoom guards against a missing coordinator (never-started socket), matching the defensive pattern of the readiness getters. - reconnect-coordinator.ts: handleDisconnect emits 'disconnected' at most once per outage (guard against a redundant disconnect while already disconnected). - tests: cover the raw-joinId echo and the single-disconnect-per-outage guard. --- spa/server.js | 4 +++- spa/src/reconnect-coordinator.ts | 5 ++++- spa/src/socket.ts | 1 + spa/tests/reconnect-coordinator.test.ts | 12 ++++++++++++ spa/tests/server-presence.test.ts | 10 ++++++++++ 5 files changed, 30 insertions(+), 2 deletions(-) diff --git a/spa/server.js b/spa/server.js index f4c2e80a..d0808f57 100644 --- a/spa/server.js +++ b/spa/server.js @@ -109,7 +109,9 @@ io.on("connection", async function(socket) { // a stale/superseded reply can never settle a newer client attempt. if (typeof joinId !== "string" || joinId.length === 0 || joinId.length > MAX_ID_LENGTH) { console.error("JOIN has invalid joinId!"); - socket.emit("JOIN:error", { room, joinId: undefined, reason: "invalid_join_id" }); + // Echo the client's raw joinId back (not undefined) so the client can + // still correlate and fail this attempt fast instead of timing out. + socket.emit("JOIN:error", { room, joinId: data.joinId, reason: "invalid_join_id" }); return; } if (room === undefined || room === null || `${room}`.length === 0) { diff --git a/spa/src/reconnect-coordinator.ts b/spa/src/reconnect-coordinator.ts index cd55dd28..48c53ad4 100644 --- a/spa/src/reconnect-coordinator.ts +++ b/spa/src/reconnect-coordinator.ts @@ -202,6 +202,7 @@ export class ReconnectCoordinator { /** The transport dropped. Keep the logical intent pending for auto-rejoin. */ public handleDisconnect(): void { + const wasDisconnected = this.phaseValue === "disconnected"; // Abort the in-flight attempt WITHOUT rejecting the logical intent - the // reconnect will start a fresh attempt for the same desired room. this.abortCurrentAttempt("disconnected"); @@ -211,7 +212,9 @@ export class ReconnectCoordinator { } else { this.phaseValue = "idle"; } - this.emitLifecycle("disconnected"); + // Exactly one "disconnected" per outage: a redundant disconnect while + // already disconnected must not re-emit to lifecycle consumers. + if (!wasDisconnected) this.emitLifecycle("disconnected"); } // ---- internals ---- diff --git a/spa/src/socket.ts b/spa/src/socket.ts index 7d39218d..9097da8c 100644 --- a/spa/src/socket.ts +++ b/spa/src/socket.ts @@ -129,6 +129,7 @@ class SocketManager { * @param roomId id of room to leave */ public leaveRoom(roomId: string|number): void { + if (!this.coordinator) return; // never started - nothing to leave const wasConnected = this.connected; this.coordinator.clearRoomIntent(roomId); if (wasConnected) this.socket.emit("unsubscribe", { room: roomId }); diff --git a/spa/tests/reconnect-coordinator.test.ts b/spa/tests/reconnect-coordinator.test.ts index b1660368..4b1fe954 100644 --- a/spa/tests/reconnect-coordinator.test.ts +++ b/spa/tests/reconnect-coordinator.test.ts @@ -292,6 +292,18 @@ test("lifecycle unsubscribe stops delivery and subscriptions do not accumulate", void p; }); +test("a redundant disconnect while already disconnected does not re-emit disconnected", async () => { + const s = setup(); + const p = track(s.coord.requestRoom("room-A", "token")); + s.roomState("room-A", s.lastJoin().joinId); + await tick(); + + s.coord.handleDisconnect(); + s.coord.handleDisconnect(); // redundant - must not double-fire + assert.strictEqual(s.events.filter(e => e === "disconnected").length, 1); + void p; +}); + async function run(): Promise { let failures = 0; for (const { name, run: runTest } of tests) { diff --git a/spa/tests/server-presence.test.ts b/spa/tests/server-presence.test.ts index b188d7b4..9dffbfe7 100644 --- a/spa/tests/server-presence.test.ts +++ b/spa/tests/server-presence.test.ts @@ -177,6 +177,16 @@ test("a malformed JOIN payload is rejected without crashing the server", async ( assert.strictEqual(rs.room, "room-alive"); }); +test("an invalid joinId JOIN:error echoes the client's raw joinId so it stays correlatable", async () => { + const sock = await newClient(); + const token = signToken(70, "grace"); + const err = waitFor(sock, "JOIN:error", (p) => p.reason === "invalid_join_id"); + sock.emit("JOIN", { room: "room-jid", token, presenceId: "pres-j", joinId: "" }); // empty => invalid + const payload = await err; + assert.strictEqual(payload.joinId, ""); // echoed back, not undefined + assert.strictEqual(payload.room, "room-jid"); +}); + test("a duplicate same-room JOIN does not re-announce the presence to peers", async () => { const room = "room-dup"; const peer = await newClient(); From 016399111aa5601346e240c53a572ef1323b369b Mon Sep 17 00:00:00 2001 From: Ryan Bundy Date: Tue, 21 Jul 2026 07:27:08 -0400 Subject: [PATCH 5/6] fix: recover stationary avatar position on reconnect (viewpoint resend) Browser QA of the socket-service-restart scenario surfaced a real defect: after reconnect, a stationary 3D user's position was never restored on peers (lasting origin/no-position), violating the A6 recovery goal. Root cause + fixes (narrow, reconnect/resync only): - sendInitialViewpoint emitted the transform under ({detail:{pos,rot}}), but the server (msg.pos/msg.rot) and onPresenceMoved (event.pos/event.rot) read TOP-LEVEL only, so the viewpoint resend was silently ignored by peers and never stored server-side. Add presence.avTransformPayload(pos,rot) -> {pos,rot} and send it top-level, matching the movement watchers. - The one-shot resync viewpoint went through sendAv's volatile emit, so a busy or backgrounded reconnecting tab could drop it. Add sendAv(payload,{reliable}) and send the recovery viewpoint reliably; high-frequency movement stays volatile. Tests: avTransformPayload shape unit test; a server-presence protocol test proving a top-level-pos viewpoint reaches peers and is stored for late joiners. 53/53 pass, build clean. Verified in-browser: stationary peer position now recovers after a socket restart even when the reconnecting sender is a backgrounded tab. --- .../pages/world-browser/WorldBrowserPage.vue | 26 +++++++------------ spa/src/presence.ts | 13 ++++++++++ spa/src/socket.ts | 17 ++++++++---- spa/tests/presence.test.ts | 15 ++++++++++- spa/tests/server-presence.test.ts | 24 +++++++++++++++++ 5 files changed, 73 insertions(+), 22 deletions(-) diff --git a/spa/src/pages/world-browser/WorldBrowserPage.vue b/spa/src/pages/world-browser/WorldBrowserPage.vue index be98fea8..4fb646f8 100644 --- a/spa/src/pages/world-browser/WorldBrowserPage.vue +++ b/spa/src/pages/world-browser/WorldBrowserPage.vue @@ -43,7 +43,7 @@ import { debugMsg, environment, } from "@/helpers"; -import { PresenceStore, Presence, presenceKey, isSelfPresence, isPresenceEventForRoom } from "@/presence"; +import { PresenceStore, Presence, presenceKey, isSelfPresence, isPresenceEventForRoom, avTransformPayload } from "@/presence"; import { WorldBrowserData } from "./world-browser-data.interface"; export default Vue.extend({ @@ -316,21 +316,15 @@ export default Vue.extend({ sendInitialViewpoint(): void { if(this.$store.data.view3d){ const { viewpointPosition, viewpointOrientation } = X3D.getBrowser(this.browser); - this.$socket.sendAv({ - detail: { - pos: [ - viewpointPosition.x, - viewpointPosition.y, - viewpointPosition.z, - ], - rot: [ - viewpointOrientation.x, - viewpointOrientation.y, - viewpointOrientation.z, - viewpointOrientation.angle, - ], - }, - }); + // Emit the transform TOP-LEVEL (pos/rot), matching the movement watchers + // and what the server (`msg.pos`/`msg.rot`) and peers (`onPresenceMoved`) + // actually consume. A `detail`-wrapped payload is silently ignored by + // both, so a reconnect-driven resend would never restore a stationary + // user's position on peers (they'd wait for real movement). + this.$socket.sendAv(avTransformPayload( + [viewpointPosition.x, viewpointPosition.y, viewpointPosition.z], + [viewpointOrientation.x, viewpointOrientation.y, viewpointOrientation.z, viewpointOrientation.angle], + ), { reliable: true }); // one-shot recovery packet - must not be dropped } }, moveObject(objectId): void { diff --git a/spa/src/presence.ts b/spa/src/presence.ts index 630eaf4a..46454154 100644 --- a/spa/src/presence.ts +++ b/spa/src/presence.ts @@ -65,6 +65,19 @@ export function isPresenceEventForRoom( return `${eventRoom}` === `${activeRoom}`; } +/** + * Builds an AV movement/viewpoint payload with the transform at the TOP LEVEL + * (`pos`/`rot`) - the shape the server (`msg.pos`/`msg.rot`) and the client + * (`onPresenceMoved` reading `event.pos`/`event.rot`) both consume. Nesting the + * transform under `detail` makes it silently ignored by both, which is exactly + * what stopped a reconnect viewpoint-resend from restoring a stationary user's + * position on peers. Keep viewpoint resends going through here so they stay in + * lockstep with the movement watchers. + */ +export function avTransformPayload(pos: Position3, rot: Rotation4): { pos: Position3; rot: Rotation4 } { + return { pos, rot }; +} + export interface ReconcileResult { added: Presence[]; updated: Presence[]; diff --git a/spa/src/socket.ts b/spa/src/socket.ts index 9097da8c..7c090a88 100644 --- a/spa/src/socket.ts +++ b/spa/src/socket.ts @@ -94,16 +94,23 @@ class SocketManager { /** * Emits a room-scoped AV (avatar movement/gesture/viewpoint) payload. Dropped * entirely while disconnected so nothing is buffered by Socket.IO and flushed - * into a later room, and volatile so stale movement is never queued. The - * authoritative current room is stamped on so the server can reject any AV - * that doesn't match the socket's current room. + * into a later room. The authoritative current room is stamped on so the + * server can reject any AV that doesn't match the socket's current room. + * + * High-frequency movement is sent `volatile` (a dropped frame is harmless and + * must never queue), but a one-shot critical send (`opts.reliable`) - notably + * the post-reconnect viewpoint resend that restores a stationary user's + * position on peers - must NOT be volatile, or a busy/backgrounded tab can + * silently drop it and peers would keep the user at the origin. * @param payload the AV detail to broadcast + * @param opts `reliable: true` for a one-shot that must not be dropped */ - public sendAv(payload: Record): void { + public sendAv(payload: Record, opts: { reliable?: boolean } = {}): void { if (!this.socket || !this.socket.connected) return; // drop while disconnected const room = this.coordinator ? this.coordinator.currentRoom : null; if (room == null) return; - this.socket.volatile.emit("AV", { ...payload, room }); + const channel = opts.reliable ? this.socket : this.socket.volatile; + channel.emit("AV", { ...payload, room }); } /** diff --git a/spa/tests/presence.test.ts b/spa/tests/presence.test.ts index a527e09c..fd494477 100644 --- a/spa/tests/presence.test.ts +++ b/spa/tests/presence.test.ts @@ -8,7 +8,7 @@ */ import assert from "assert"; import { EventEmitter } from "events"; -import { PresenceStore, presenceKey, isSelfPresence, isPresenceEventForRoom, Presence } from "../src/presence"; +import { PresenceStore, presenceKey, isSelfPresence, isPresenceEventForRoom, avTransformPayload, Presence } from "../src/presence"; import { joinRoomOverSocket } from "../src/join-protocol"; type Test = { name: string; run: () => void | Promise }; @@ -302,6 +302,19 @@ describe("isPresenceEventForRoom", () => { }); }); +describe("avTransformPayload", () => { + test("puts pos/rot at the TOP level (never nested under detail)", () => { + const payload = avTransformPayload([1, 2, 3], [0, 1, 0, 0.5]); + // The server (msg.pos/msg.rot) and onPresenceMoved (event.pos/event.rot) + // read top-level only; a `detail`-wrapped transform is silently dropped and + // would break reconnect viewpoint recovery. + assert.deepStrictEqual(payload, { pos: [1, 2, 3], rot: [0, 1, 0, 0.5] }); + assert.strictEqual((payload as any).detail, undefined); + assert.strictEqual(Object.prototype.hasOwnProperty.call(payload, "pos"), true); + assert.strictEqual(Object.prototype.hasOwnProperty.call(payload, "rot"), true); + }); +}); + async function run(): Promise { let failures = 0; for (const { name, run: runTest } of tests) { diff --git a/spa/tests/server-presence.test.ts b/spa/tests/server-presence.test.ts index 9dffbfe7..b729e972 100644 --- a/spa/tests/server-presence.test.ts +++ b/spa/tests/server-presence.test.ts @@ -267,6 +267,30 @@ test("a stale old socket cannot delete or announce a presence a newer socket now assert.strictEqual(stillThere, true); }); +test("a top-level-pos viewpoint resend updates a stationary user's position for peers and late joiners", async () => { + // Mirrors the reconnect A6 recovery: a stationary user re-announces its + // viewpoint (top-level pos/rot, as avTransformPayload produces). Peers must + // apply it AND the server must store it, so a peer that reconnects/joins + // afterwards sees the real position via ROOM_STATE - not the origin. + const room = "room-viewpoint"; + const peer = await newClient(); + await join(peer, room, signToken(80, "peer80"), "pres-peer80", "jid-vp1"); + const subject = await newClient(); + await join(subject, room, signToken(81, "grace81"), "pres-vp", "jid-vp2"); + + // Stationary viewpoint resend (never nested under `detail`). + const gotAv = waitFor(peer, "AV", (p) => p.presenceId === "pres-vp"); + subject.emit("AV", { room, pos: [12, 3, 45], rot: [0, 1, 0, 1.5] }); + const av = await gotAv; + assert.deepStrictEqual(av.pos, [12, 3, 45]); // peer applied the position + + const late = await newClient(); + const rs = await join(late, room, signToken(82, "late82"), "pres-late", "jid-vp3"); + const stored = rs.presences.find((p: any) => p.presenceId === "pres-vp"); + assert.ok(stored, "stationary user present in snapshot"); + assert.deepStrictEqual(stored.pos, [12, 3, 45]); // server stored it for late joiners +}); + test("a stale old socket cannot broadcast AV under a presence a newer socket owns", async () => { const room = "room-staleav"; const peer = await newClient(); From a1bae5a220446726acb869d1f5f1f221cb60cf29 Mon Sep 17 00:00:00 2001 From: Ryan Bundy Date: Thu, 30 Jul 2026 11:14:27 -0400 Subject: [PATCH 6/6] Enable chat on the first join, not only on a resync Copilot review of #5. onSocketLifecycle handled "disconnected" and "resynced" and ignored "ready". But the coordinator emits `recovered ? "resynced" : "ready"`, so a FIRST successful join only ever emits "ready" -- "resynced" means a rejoin that recovered an existing room. mounted() enables chat only when $socket.roomReady is already true, so a Chat that mounted before the room came up had no path to enabled at all: input stayed hidden behind v-show="chatEnabled", and startNewChat, canAdmin, getRole, getXpAmount, joinedChat and the timers never ran. Whether it broke depended purely on whether the join beat the mount, which is why it would present as chat intermittently not working rather than as a clean failure. The comment above the mounted() check asserted "onSocketLifecycle flips it on the resynced transition if we mounted mid-resync", which is precisely the wrong assumption. The mount-time bundle is now activateRoom(), called from mounted() when the room is already ready and from the "ready" event when it was not. Guarded by roomActivated so both paths cannot double-run: startNewChat clears this.messages, and re-running it would wipe the visible history. That is also why "ready" on an already-activated Chat -- a rejoin that could not be resynced -- restores input without calling activateRoom, matching what the "resynced" branch has always been careful to do. "failed" is now handled too. Without it the user keeps the "Reconnecting to chat server..." line from the disconnect forever, which stops being true the moment the coordinator gives up. Beyond the finding, but leaving a message that has become a lie is not a state worth preserving. socket.ts: start() returns Promise.resolve() on the already-started path instead of a bare `return`. The signature promises a Promise, so it was handing back undefined and any caller chaining .then() would throw. The one current caller awaits, and `await undefined` is fine, which is why this survived -- not why it was safe. joinRoom and onLifecycle now guard this.coordinator, which only exists after start(). leaveRoom, roomReady, lifecyclePhase, pendingJoinId, currentRoom and sendAv already did. onLifecycle logs an error rather than silently returning a no-op unsubscribe: a dropped subscription means the subscriber never learns the room came up, which is the same failure this commit is fixing, and it should not be silent. Verified: 53/53 spa tests pass (29 presence, 13 reconnect-coordinator, 11 server-presence). eslint compared against a stashed baseline error-class by error-class: identical, none introduced -- two doublequote violations I added were fixed before this commit. Deliberately not done: no test covers the mount-before-ready ordering. The existing suite drives the coordinator directly and does not mount Chat, so asserting this needs component-level mounting that the harness does not currently do. Worth adding, but it is a test-infrastructure change rather than part of this fix. --- spa/src/components/Chat.vue | 61 +++++++++++++++++++++++++++++++------ spa/src/socket.ts | 21 ++++++++++++- 2 files changed, 72 insertions(+), 10 deletions(-) diff --git a/spa/src/components/Chat.vue b/spa/src/components/Chat.vue index e4710978..fbdb3ae6 100644 --- a/spa/src/components/Chat.vue +++ b/spa/src/components/Chat.vue @@ -306,6 +306,8 @@ interface ChatData { pingIntervalId: any; worldMembers: any[]; chatEnabled: boolean; + /** Whether the one-time room activation has run for this Chat instance. */ + roomActivated: boolean; unsubscribeLifecycle: (() => void) | null; showRole: boolean; showXP: boolean; @@ -404,6 +406,7 @@ export default Vue.extend