diff --git a/src/components/panels/ConnectionStatusBar.tsx b/src/components/panels/ConnectionStatusBar.tsx new file mode 100644 index 0000000..81c3341 --- /dev/null +++ b/src/components/panels/ConnectionStatusBar.tsx @@ -0,0 +1,110 @@ +"use client"; + +import { useConnectionState } from "@/store/slices/connectionSlice"; +import { + MAX_RECONNECT_ATTEMPTS, + type ConnectionQuality, + type ConnectionStatus, +} from "@/types/connection"; + +/** + * Fixed top bar showing live connection quality (green / yellow / red). It + * reads {@link useConnectionState} and surfaces reconnection progress, plus a + * persistent error banner once the reconnection stratum gives up. + */ + +const QUALITY_STYLES: Record< + ConnectionQuality, + { dot: string; bar: string; label: string } +> = { + good: { dot: "bg-green-500", bar: "bg-green-500/10 text-green-600", label: "Connected" }, + degraded: { + dot: "bg-amber-500 animate-pulse", + bar: "bg-amber-500/10 text-amber-600", + label: "Unstable", + }, + down: { dot: "bg-red-500", bar: "bg-red-500/10 text-red-600", label: "Disconnected" }, +}; + +function statusLabel(status: ConnectionStatus, attempt: number): string { + switch (status) { + case "connected": + return "Live"; + case "connecting": + return "Connecting…"; + case "reconnecting": + return `Reconnecting (attempt ${attempt}/${MAX_RECONNECT_ATTEMPTS})…`; + case "recovering": + return "Recovering subscription…"; + case "failed": + return "Connection lost"; + default: + return "Idle"; + } +} + +export interface ConnectionStatusBarProps { + /** Optional retry handler wired to the failure banner. */ + onRetry?: () => void; + className?: string; +} + +export function ConnectionStatusBar({ onRetry, className }: ConnectionStatusBarProps) { + const { status, quality, latency, nodeId, attempt, lastError } = + useConnectionState(); + + const styles = QUALITY_STYLES[quality]; + const failed = status === "failed"; + + return ( +
+
+
+
+
+ {status === "connected" && latency >= 0 && ( + {latency === 0 ? "stable" : `${latency} ms drift`} + )} + Link quality: {styles.label} +
+
+ + {failed && ( +
+ + {lastError ?? + `Reconnection failed after ${MAX_RECONNECT_ATTEMPTS} attempts.`}{" "} + Telemetry is paused. + + {onRetry && ( + + )} +
+ )} +
+ ); +} + +export default ConnectionStatusBar; diff --git a/src/hooks/useWebSocket.ts b/src/hooks/useWebSocket.ts new file mode 100644 index 0000000..6114eba --- /dev/null +++ b/src/hooks/useWebSocket.ts @@ -0,0 +1,447 @@ +"use client"; + +import { useCallback, useEffect, useRef, useState } from "react"; +import type { SocketLike, SocketFactory } from "@/hooks/useTelemetryStream"; +import { + HEARTBEAT_INTERVAL_MS, + HEARTBEAT_TIMEOUT_MS, + STICKY_NODE_STORAGE_KEY, + type ConnectionStatus, + type InboundControl, + type RecoveryAck, + type RecoveryFrame, + type TelemetryFrame, +} from "@/types/connection"; +import { ReconnectMachine } from "@/services/reconnectMachine"; +import { FrameBuffer } from "@/utils/frameBuffer"; +import { connectionStore } from "@/store/slices/connectionSlice"; + +type TimerId = ReturnType; + +export interface UseWebSocketOptions { + /** Base WebSocket URL; `?nodeId={sticky}` is appended automatically. */ + url: string; + /** Start the connection. @default true */ + enabled?: boolean; + /** Injectable transport (defaults to the browser WebSocket). */ + createSocket?: SocketFactory; + /** Called for every inbound telemetry frame (live and backfilled). */ + onFrame?: (frame: TelemetryFrame) => void; + /** + * REST backfill for frames missed during the outage. Defaults to + * `GET /ws/replay?from=&to=`. Injectable for tests. + */ + replayFetch?: (from: number, to: number) => Promise; + /** Injectable clock (ms). @default Date.now */ + now?: () => number; + /** Injectable RNG for backoff jitter. @default Math.random */ + rng?: () => number; + /** Injectable session storage for the sticky node id. */ + storage?: Pick; +} + +export interface UseWebSocketResult { + status: ConnectionStatus; + attempt: number; + missedHeartbeats: number; + nodeId: string | null; + lastError: string | null; + /** Send an application message over the socket (no-op if not open). */ + send: (data: unknown) => boolean; + connect: () => void; + disconnect: () => void; +} + +function defaultReplayFetch( + from: number, + to: number +): Promise { + return fetch(`/ws/replay?from=${from}&to=${to}`, { + headers: { Accept: "application/json" }, + }).then((res) => { + if (!res.ok) throw new Error(`Replay failed: HTTP ${res.status}`); + return res.json() as Promise; + }); +} + +function getDefaultStorage(): UseWebSocketOptions["storage"] | undefined { + try { + return typeof sessionStorage !== "undefined" ? sessionStorage : undefined; + } catch { + return undefined; // access can throw in sandboxed iframes + } +} + +function isControl(msg: unknown): msg is InboundControl { + return ( + typeof msg === "object" && + msg !== null && + typeof (msg as { type?: unknown }).type === "string" + ); +} + +/** + * Persistent WebSocket with adaptive reconnection. Connection lifecycle is + * delegated to {@link ReconnectMachine}; on entering `reconnecting` the hook + * arms a full-jitter backoff timer, and on reconnect it runs a subscription + * recovery handshake (replaying the buffered frames) before resuming. Status + * transitions are mirrored into {@link connectionStore} for the UI. + */ +export function useWebSocket(options: UseWebSocketOptions): UseWebSocketResult { + const { + url, + enabled = true, + createSocket, + onFrame, + replayFetch = defaultReplayFetch, + now = Date.now, + rng = Math.random, + storage = getDefaultStorage(), + } = options; + + const [status, setStatus] = useState("idle"); + const [attempt, setAttempt] = useState(0); + const [missedHeartbeats, setMissedHeartbeats] = useState(0); + const [nodeId, setNodeId] = useState(null); + const [lastError, setLastError] = useState(null); + + // Stable singletons for the connection lifetime. + const machineRef = useRef(null); + if (machineRef.current === null) { + machineRef.current = new ReconnectMachine({ rng }); + } + const bufferRef = useRef(null); + if (bufferRef.current === null) { + bufferRef.current = new FrameBuffer(); + } + + const socketRef = useRef(null); + const reconnectTimerRef = useRef(null); + const heartbeatTimerRef = useRef(null); + const lastPingAtRef = useRef(0); + const disposedRef = useRef(false); + + // Latest callbacks/config without re-subscribing the socket. + const onFrameRef = useRef(onFrame); + onFrameRef.current = onFrame; + const createSocketRef = useRef(createSocket); + createSocketRef.current = createSocket; + const replayFetchRef = useRef(replayFetch); + replayFetchRef.current = replayFetch; + + // ---- helpers ------------------------------------------------------------ + + const clearReconnectTimer = useCallback(() => { + if (reconnectTimerRef.current !== null) { + clearTimeout(reconnectTimerRef.current); + reconnectTimerRef.current = null; + } + }, []); + + const clearHeartbeatTimer = useCallback(() => { + if (heartbeatTimerRef.current !== null) { + clearTimeout(heartbeatTimerRef.current); + heartbeatTimerRef.current = null; + } + }, []); + + /** Mirror machine context into local state and the global store. */ + const syncFromMachine = useCallback(() => { + const ctx = machineRef.current!.getContext(); + setStatus(ctx.status); + setAttempt(ctx.attempt); + setMissedHeartbeats(ctx.missedHeartbeats); + setLastError(ctx.lastError); + if (ctx.status === "failed") { + connectionStore.dispatch({ + type: "CONNECTION_FAILED", + payload: { error: ctx.lastError ?? "Connection failed" }, + }); + } else { + connectionStore.dispatch({ + type: "CONNECTION_STATUS_CHANGED", + payload: { + status: ctx.status, + attempt: ctx.attempt, + nextDelayMs: ctx.nextDelayMs, + }, + }); + } + }, []); + + const buildUrl = useCallback((): string => { + const sticky = storage?.getItem(STICKY_NODE_STORAGE_KEY) ?? ""; + const sep = url.includes("?") ? "&" : "?"; + return sticky ? `${url}${sep}nodeId=${encodeURIComponent(sticky)}` : url; + }, [storage, url]); + + // Forward declaration so the close handler can re-open. + const openRef = useRef<() => void>(() => {}); + + const scheduleReconnect = useCallback(() => { + const ctx = machineRef.current!.getContext(); + if (ctx.status !== "reconnecting" || ctx.nextDelayMs === null) return; + clearReconnectTimer(); + reconnectTimerRef.current = setTimeout(() => { + reconnectTimerRef.current = null; + if (disposedRef.current) return; + machineRef.current!.send({ type: "RETRY" }); + syncFromMachine(); + openRef.current(); + }, ctx.nextDelayMs); + }, [clearReconnectTimer, syncFromMachine]); + + const handleDisconnect = useCallback(() => { + if (disposedRef.current) return; + clearHeartbeatTimer(); + socketRef.current = null; + machineRef.current!.send({ type: "DISCONNECTED" }); + syncFromMachine(); + if (machineRef.current!.getContext().status === "reconnecting") { + scheduleReconnect(); + } + }, [clearHeartbeatTimer, scheduleReconnect, syncFromMachine]); + + /** Arm the heartbeat watchdog: a missed server ping marks the link dead. */ + const armHeartbeat = useCallback(() => { + clearHeartbeatTimer(); + heartbeatTimerRef.current = setTimeout(() => { + if (disposedRef.current) return; + const missed = machineRef.current!.recordMissedHeartbeat(); + connectionStore.dispatch({ + type: "HEARTBEAT_MISSED", + payload: { missedHeartbeats: missed }, + }); + setMissedHeartbeats(missed); + machineRef.current!.send({ type: "HEARTBEAT_TIMEOUT" }); + syncFromMachine(); + try { + socketRef.current?.close(); + } catch { + /* ignore */ + } + handleDisconnect(); + }, HEARTBEAT_INTERVAL_MS + HEARTBEAT_TIMEOUT_MS); + }, [clearHeartbeatTimer, handleDisconnect, syncFromMachine]); + + const handlePing = useCallback(() => { + const t = now(); + const prev = lastPingAtRef.current; + lastPingAtRef.current = t; + // Inter-ping drift beyond the nominal interval is reported as latency. + if (prev > 0) { + const latency = Math.max(0, t - prev - HEARTBEAT_INTERVAL_MS); + connectionStore.dispatch({ type: "HEARTBEAT_LATENCY", payload: { latency } }); + } + try { + socketRef.current?.send(JSON.stringify({ type: "pong" })); + } catch { + /* ignore */ + } + armHeartbeat(); + }, [armHeartbeat, now]); + + const sendRecovery = useCallback(() => { + const buffer = bufferRef.current!; + const sticky = storage?.getItem(STICKY_NODE_STORAGE_KEY) ?? null; + const frame: RecoveryFrame = { + type: "recovery", + lastSequenceId: buffer.lastSequenceId, + nodeId: sticky, + frames: buffer.drain(), + }; + try { + socketRef.current?.send(JSON.stringify(frame)); + } catch { + /* ignore */ + } + }, [storage]); + + const handleRecoveryAck = useCallback( + async (ack: RecoveryAck) => { + const buffer = bufferRef.current!; + const fromSeq = buffer.lastSequenceId; + + machineRef.current!.send({ type: "RECOVERY_SUCCESS" }); + syncFromMachine(); + connectionStore.dispatch({ type: "CONNECTION_RECOVERED" }); + armHeartbeat(); + + // Backfill anything the buffer could not cover via the REST replay API. + if (ack.missedCount > 0 && ack.serverCurrentSeq > fromSeq) { + try { + const missed = await replayFetchRef.current(fromSeq, ack.serverCurrentSeq); + if (disposedRef.current) return; + for (const f of missed) { + buffer.push(f); + onFrameRef.current?.(f); + } + } catch { + // Backfill is best-effort; the live stream continues regardless. + } + } + }, + [armHeartbeat, syncFromMachine] + ); + + const handleMessage = useCallback( + (raw: unknown) => { + let msg: unknown = raw; + if (typeof raw === "string") { + try { + msg = JSON.parse(raw); + } catch { + return; // ignore non-JSON text frames + } + } + + if (isControl(msg)) { + switch (msg.type) { + case "ping": + handlePing(); + return; + case "node_assigned": + storage?.setItem(STICKY_NODE_STORAGE_KEY, msg.nodeId); + setNodeId(msg.nodeId); + connectionStore.dispatch({ + type: "NODE_ID_ASSIGNED", + payload: { nodeId: msg.nodeId }, + }); + return; + case "recovery_ack": + void handleRecoveryAck(msg); + return; + } + } + + // Otherwise treat it as a telemetry frame and buffer it for replay. + if ( + typeof msg === "object" && + msg !== null && + typeof (msg as { sequenceId?: unknown }).sequenceId === "number" + ) { + const frame = msg as TelemetryFrame; + bufferRef.current!.push(frame); + onFrameRef.current?.(frame); + } + }, + [handlePing, handleRecoveryAck, storage] + ); + + const open = useCallback(() => { + if (disposedRef.current) return; + let socket: SocketLike; + try { + const factory = createSocketRef.current; + if (!factory) throw new Error("No socket factory provided"); + socket = factory(buildUrl()); + } catch (err) { + machineRef.current!.send({ type: "DISCONNECTED" }); + setLastError((err as Error).message); + syncFromMachine(); + if (machineRef.current!.getContext().status === "reconnecting") { + scheduleReconnect(); + } + return; + } + socketRef.current = socket; + + socket.onopen = () => { + if (disposedRef.current) return; + machineRef.current!.send({ type: "CONNECTED" }); + syncFromMachine(); + const ctx = machineRef.current!.getContext(); + if (ctx.status === "recovering") { + sendRecovery(); + } else { + armHeartbeat(); + } + }; + socket.onmessage = (ev: MessageEvent) => { + if (disposedRef.current) return; + handleMessage(ev.data); + }; + socket.onclose = () => handleDisconnect(); + socket.onerror = () => handleDisconnect(); + }, [ + armHeartbeat, + buildUrl, + handleDisconnect, + handleMessage, + scheduleReconnect, + sendRecovery, + syncFromMachine, + ]); + openRef.current = open; + + // ---- public API --------------------------------------------------------- + + const connect = useCallback(() => { + if (disposedRef.current) return; + machineRef.current!.send({ type: "CONNECT" }); + syncFromMachine(); + open(); + }, [open, syncFromMachine]); + + const disconnect = useCallback(() => { + clearReconnectTimer(); + clearHeartbeatTimer(); + machineRef.current!.send({ type: "RESET" }); + syncFromMachine(); + connectionStore.dispatch({ type: "RESET" }); + const socket = socketRef.current; + socketRef.current = null; + if (socket) { + try { + socket.close(); + } catch { + /* ignore */ + } + } + }, [clearHeartbeatTimer, clearReconnectTimer, syncFromMachine]); + + const send = useCallback((data: unknown): boolean => { + const socket = socketRef.current; + if (!socket || socket.readyState !== 1 /* OPEN */) return false; + try { + socket.send(typeof data === "string" ? data : JSON.stringify(data)); + return true; + } catch { + return false; + } + }, []); + + // ---- lifecycle ---------------------------------------------------------- + + useEffect(() => { + if (!enabled) return; + disposedRef.current = false; + connect(); + return () => { + disposedRef.current = true; + clearReconnectTimer(); + clearHeartbeatTimer(); + const socket = socketRef.current; + socketRef.current = null; + if (socket) { + try { + socket.close(); + } catch { + /* ignore */ + } + } + }; + // Intentionally only (re)run when enabled/url changes; callbacks use refs. + }, [enabled, url]); + + return { + status, + attempt, + missedHeartbeats, + nodeId, + lastError, + send, + connect, + disconnect, + }; +} diff --git a/src/services/reconnectMachine.ts b/src/services/reconnectMachine.ts new file mode 100644 index 0000000..a334289 --- /dev/null +++ b/src/services/reconnectMachine.ts @@ -0,0 +1,201 @@ +/** + * Custom finite state machine governing the WebSocket reconnection lifecycle. + * + * States: idle → connecting → connected, with reconnecting / recovering / + * failed branches. The machine is pure: it computes the next backoff delay and + * exposes it via context, but it schedules nothing itself — the owning hook + * reads {@link ReconnectContext.nextDelayMs}, arms a timer, and feeds a `RETRY` + * event back in. This keeps the transition logic deterministic and testable. + */ + +import { + MAX_RECONNECT_ATTEMPTS, + type ConnectionStatus, + type ReconnectContext, + type ReconnectEvent, +} from "@/types/connection"; +import { fullJitterBackoff } from "@/utils/backoff"; + +export interface ReconnectMachineOptions { + /** Injectable RNG for jitter (defaults to Math.random) — used in tests. */ + rng?: () => number; + /** Override the terminal-failure attempt threshold. */ + maxAttempts?: number; +} + +type Listener = (context: ReconnectContext) => void; + +const INITIAL_CONTEXT: ReconnectContext = { + status: "idle", + attempt: 0, + missedHeartbeats: 0, + nextDelayMs: null, + lastError: null, +}; + +export class ReconnectMachine { + private context: ReconnectContext = { ...INITIAL_CONTEXT }; + private readonly listeners = new Set(); + private readonly rng: () => number; + private readonly maxAttempts: number; + + constructor(options: ReconnectMachineOptions = {}) { + this.rng = options.rng ?? Math.random; + this.maxAttempts = options.maxAttempts ?? MAX_RECONNECT_ATTEMPTS; + } + + getContext(): Readonly { + return this.context; + } + + get status(): ConnectionStatus { + return this.context.status; + } + + get isTerminal(): boolean { + return this.context.status === "failed"; + } + + subscribe(listener: Listener): () => void { + this.listeners.add(listener); + return () => this.listeners.delete(listener); + } + + /** Feed an event into the machine, applying the transition and notifying. */ + send(event: ReconnectEvent): ReconnectContext { + const next = this.transition(this.context, event); + if (next !== this.context) { + this.context = next; + this.notify(); + } + return this.context; + } + + /** + * Record a missed heartbeat. Returns the running count so the caller can + * decide whether to escalate to a `HEARTBEAT_TIMEOUT` event. + */ + recordMissedHeartbeat(): number { + this.context = { + ...this.context, + missedHeartbeats: this.context.missedHeartbeats + 1, + }; + this.notify(); + return this.context.missedHeartbeats; + } + + private transition( + ctx: ReconnectContext, + event: ReconnectEvent + ): ReconnectContext { + // RESET is accepted from any state. + if (event.type === "RESET") { + return { ...INITIAL_CONTEXT }; + } + + switch (ctx.status) { + case "idle": + if (event.type === "CONNECT") { + return { ...INITIAL_CONTEXT, status: "connecting" }; + } + return ctx; + + case "connecting": + switch (event.type) { + case "CONNECTED": + // A successful open after an outage enters recovery; the very first + // connection has nothing to recover and goes straight to connected. + return ctx.attempt > 0 + ? { ...ctx, status: "recovering", nextDelayMs: null } + : { + ...ctx, + status: "connected", + missedHeartbeats: 0, + nextDelayMs: null, + lastError: null, + }; + case "DISCONNECTED": + case "HEARTBEAT_TIMEOUT": + return this.enterReconnecting(ctx, event); + default: + return ctx; + } + + case "connected": + if (event.type === "DISCONNECTED" || event.type === "HEARTBEAT_TIMEOUT") { + return this.enterReconnecting(ctx, event); + } + return ctx; + + case "reconnecting": + if (event.type === "RETRY") { + return { ...ctx, status: "connecting", nextDelayMs: null }; + } + // Already reconnecting; ignore duplicate disconnect signals. + return ctx; + + case "recovering": + switch (event.type) { + case "RECOVERY_SUCCESS": + return { + ...ctx, + status: "connected", + attempt: 0, + missedHeartbeats: 0, + nextDelayMs: null, + lastError: null, + }; + case "DISCONNECTED": + case "HEARTBEAT_TIMEOUT": + return this.enterReconnecting(ctx, event); + default: + return ctx; + } + + case "failed": + // Manual retry restarts the lifecycle from scratch. + if (event.type === "CONNECT") { + return { ...INITIAL_CONTEXT, status: "connecting" }; + } + return ctx; + + default: + return ctx; + } + } + + /** Increment the attempt counter and either schedule a retry or give up. */ + private enterReconnecting( + ctx: ReconnectContext, + event: ReconnectEvent + ): ReconnectContext { + const attempt = ctx.attempt + 1; + const reason = + event.type === "HEARTBEAT_TIMEOUT" + ? "Heartbeat timed out" + : "Connection lost"; + + if (attempt > this.maxAttempts) { + return { + ...ctx, + status: "failed", + nextDelayMs: null, + lastError: `Reconnection failed after ${this.maxAttempts} attempts`, + }; + } + + // Zero-based attempt index for the backoff curve (first retry → 2^0). + const nextDelayMs = fullJitterBackoff(attempt - 1, undefined, undefined, this.rng); + return { + ...ctx, + status: "reconnecting", + attempt, + nextDelayMs, + lastError: reason, + }; + } + + private notify(): void { + for (const listener of this.listeners) listener(this.context); + } +} diff --git a/src/store/slices/connectionSlice.ts b/src/store/slices/connectionSlice.ts new file mode 100644 index 0000000..c6e425b --- /dev/null +++ b/src/store/slices/connectionSlice.ts @@ -0,0 +1,167 @@ +"use client"; + +import { useSyncExternalStore } from "react"; +import { + HEARTBEAT_TIMEOUT_MS, + type ConnectionQuality, + type ConnectionStatus, +} from "@/types/connection"; + +/** + * Connection state store (custom singleton, matching the codebase's store + * pattern). UI components subscribe via {@link useConnectionState} and react to + * status transitions — e.g. dimming charts while reconnecting or showing the + * terminal-failure banner. + */ + +export interface ConnectionState { + status: ConnectionStatus; + quality: ConnectionQuality; + /** Sticky backend node id, if assigned. */ + nodeId: string | null; + /** Last measured round-trip latency in ms (-1 when unknown). */ + latency: number; + missedHeartbeats: number; + /** Current reconnection attempt (0 when healthy). */ + attempt: number; + /** Scheduled delay before the next reconnect attempt, if any. */ + nextDelayMs: number | null; + lastError: string | null; +} + +export type ConnectionAction = + | { + type: "CONNECTION_STATUS_CHANGED"; + payload: { status: ConnectionStatus; attempt?: number; nextDelayMs?: number | null }; + } + | { type: "HEARTBEAT_LATENCY"; payload: { latency: number } } + | { type: "HEARTBEAT_MISSED"; payload: { missedHeartbeats: number } } + | { type: "NODE_ID_ASSIGNED"; payload: { nodeId: string } } + | { type: "CONNECTION_FAILED"; payload: { error: string } } + | { type: "CONNECTION_RECOVERED" } + | { type: "RESET" }; + +const initialState: ConnectionState = { + status: "idle", + quality: "down", + nodeId: null, + latency: -1, + missedHeartbeats: 0, + attempt: 0, + nextDelayMs: null, + lastError: null, +}; + +/** Derive the coarse green/yellow/red quality from the detailed state. */ +export function deriveQuality( + status: ConnectionStatus, + missedHeartbeats: number, + latency: number +): ConnectionQuality { + if (status === "failed") return "down"; + if (status === "connected") { + if (missedHeartbeats > 0 || latency > HEARTBEAT_TIMEOUT_MS) return "degraded"; + return "good"; + } + if (status === "connecting" || status === "reconnecting" || status === "recovering") { + return "degraded"; + } + return "down"; // idle +} + +type Listener = (state: ConnectionState) => void; + +class ConnectionStore { + private state: ConnectionState = { ...initialState }; + private listeners = new Set(); + + getState = (): Readonly => this.state; + + subscribe = (listener: Listener): (() => void) => { + this.listeners.add(listener); + return () => this.listeners.delete(listener); + }; + + dispatch(action: ConnectionAction): void { + const next = this.reducer(this.state, action); + if (next !== this.state) { + this.state = next; + this.notify(); + } + } + + private reducer(state: ConnectionState, action: ConnectionAction): ConnectionState { + switch (action.type) { + case "CONNECTION_STATUS_CHANGED": { + const status = action.payload.status; + const attempt = action.payload.attempt ?? state.attempt; + const nextDelayMs = + action.payload.nextDelayMs !== undefined + ? action.payload.nextDelayMs + : state.nextDelayMs; + const missedHeartbeats = status === "connected" ? 0 : state.missedHeartbeats; + return { + ...state, + status, + attempt, + nextDelayMs, + missedHeartbeats, + lastError: status === "connected" ? null : state.lastError, + quality: deriveQuality(status, missedHeartbeats, state.latency), + }; + } + case "HEARTBEAT_LATENCY": + return { + ...state, + latency: action.payload.latency, + quality: deriveQuality(state.status, state.missedHeartbeats, action.payload.latency), + }; + case "HEARTBEAT_MISSED": + return { + ...state, + missedHeartbeats: action.payload.missedHeartbeats, + quality: deriveQuality(state.status, action.payload.missedHeartbeats, state.latency), + }; + case "NODE_ID_ASSIGNED": + return { ...state, nodeId: action.payload.nodeId }; + case "CONNECTION_FAILED": + return { + ...state, + status: "failed", + quality: "down", + nextDelayMs: null, + lastError: action.payload.error, + }; + case "CONNECTION_RECOVERED": + return { + ...state, + status: "connected", + attempt: 0, + missedHeartbeats: 0, + nextDelayMs: null, + lastError: null, + quality: deriveQuality("connected", 0, state.latency), + }; + case "RESET": + return { ...initialState }; + default: + return state; + } + } + + private notify(): void { + for (const listener of this.listeners) listener(this.state); + } +} + +/** Singleton connection store instance. */ +export const connectionStore = new ConnectionStore(); + +/** React binding: subscribe a component to the connection state. */ +export function useConnectionState(): ConnectionState { + return useSyncExternalStore( + connectionStore.subscribe, + connectionStore.getState, + connectionStore.getState + ); +} diff --git a/src/types/connection.ts b/src/types/connection.ts new file mode 100644 index 0000000..770d7d9 --- /dev/null +++ b/src/types/connection.ts @@ -0,0 +1,118 @@ +/** + * Types and invariants for the adaptive WebSocket reconnection stratum. + * + * The dashboard holds a persistent WebSocket to a load-balanced backend. On + * disconnect the client reconnects with capped exponential backoff + full + * jitter, reattaches to its sticky backend node, and replays buffered telemetry + * so the operator sees at most ~3 s of data loss during a failure cascade. + */ + +// --- Backoff invariants ----------------------------------------------------- + +/** Initial backoff in milliseconds. */ +export const INITIAL_BACKOFF_MS = 500; +/** Maximum backoff ceiling in milliseconds. */ +export const MAX_BACKOFF_MS = 30_000; +/** Exponential multiplier (base ^ attempt). */ +export const BACKOFF_MULTIPLIER = 2; +/** Attempts before declaring terminal failure (~3.5 min of reconnection). */ +export const MAX_RECONNECT_ATTEMPTS = 12; + +// --- Heartbeat invariants --------------------------------------------------- + +/** Server sends "ping" on this cadence (ms). */ +export const HEARTBEAT_INTERVAL_MS = 15_000; +/** Client must answer "pong" within this window or the link is dead (ms). */ +export const HEARTBEAT_TIMEOUT_MS = 5_000; + +// --- Frame buffer invariants ------------------------------------------------ + +/** Telemetry frames retained for replay during recovery. */ +export const FRAME_BUFFER_CAPACITY = 500; + +// --- Sticky session --------------------------------------------------------- + +/** sessionStorage key holding the server-assigned sticky node id. */ +export const STICKY_NODE_STORAGE_KEY = "ws:stickyNodeId"; + +// --- Connection state ------------------------------------------------------- + +/** Finite states of the reconnection machine. */ +export type ConnectionStatus = + | "idle" + | "connecting" + | "connected" + | "reconnecting" + | "recovering" + | "failed"; + +/** Coarse link quality surfaced to the UI (green / yellow / red). */ +export type ConnectionQuality = "good" | "degraded" | "down"; + +/** Events that drive the reconnection machine. */ +export type ReconnectEvent = + | { type: "CONNECT" } + | { type: "CONNECTED" } + | { type: "DISCONNECTED" } + | { type: "HEARTBEAT_TIMEOUT" } + | { type: "RETRY" } + | { type: "RECOVERY_SUCCESS" } + | { type: "RESET" }; + +/** Immutable snapshot of the machine's context. */ +export interface ReconnectContext { + status: ConnectionStatus; + /** Number of reconnection attempts made in the current outage (0 when healthy). */ + attempt: number; + /** Consecutive missed heartbeats. */ + missedHeartbeats: number; + /** Delay (ms) scheduled before the next reconnect attempt, if any. */ + nextDelayMs: number | null; + lastError: string | null; +} + +// --- Telemetry frames (replay) --------------------------------------------- + +/** + * A buffered telemetry frame. The payload is opaque to the reconnection layer; + * each frame is tagged with a monotonic sequence id so the server can compute + * how many frames the client missed during the outage. + */ +export interface TelemetryFrame { + /** Monotonic, server-assigned sequence id. */ + sequenceId: number; + /** Opaque payload (200 bytes – 100 KB). */ + data: unknown; + /** Wall-clock ms when received. */ + receivedAt: number; + /** Approximate byte size, used for buffer accounting. */ + size: number; +} + +/** Recovery handshake the client sends on reconnect. */ +export interface RecoveryFrame { + type: "recovery"; + /** Highest sequence id the client has seen. */ + lastSequenceId: number; + /** Sticky node id the client wants to reattach to. */ + nodeId: string | null; + /** Buffered frames replayed to the server as a batch. */ + frames: TelemetryFrame[]; +} + +/** Server acknowledgement of a recovery handshake. */ +export interface RecoveryAck { + type: "recovery_ack"; + /** Number of frames the client missed while disconnected. */ + missedCount: number; + /** Server's current highest sequence id (used to bound a REST backfill). */ + serverCurrentSeq: number; + /** Node id the server bound this session to (may differ from requested). */ + nodeId?: string; +} + +/** Inbound control/heartbeat messages. */ +export type InboundControl = + | { type: "ping" } + | { type: "node_assigned"; nodeId: string } + | RecoveryAck; diff --git a/src/utils/backoff.ts b/src/utils/backoff.ts new file mode 100644 index 0000000..7eeb4c9 --- /dev/null +++ b/src/utils/backoff.ts @@ -0,0 +1,48 @@ +/** + * Capped exponential backoff with full-jitter randomization. + * + * Full jitter (per AWS's "Exponential Backoff And Jitter") spreads retries + * uniformly across `[0, ceiling]` instead of clustering them at a fixed delay, + * which prevents synchronized clients from forming reconnect storms against a + * load-balanced backend. + */ + +import { + BACKOFF_MULTIPLIER, + INITIAL_BACKOFF_MS, + MAX_BACKOFF_MS, +} from "@/types/connection"; + +/** + * The deterministic ceiling for a given attempt: `min(cap, base * mult^attempt)`. + * `attempt` is zero-based (attempt 0 is the first retry). + */ +export function backoffCeiling( + attempt: number, + base: number = INITIAL_BACKOFF_MS, + cap: number = MAX_BACKOFF_MS, + multiplier: number = BACKOFF_MULTIPLIER +): number { + if (attempt < 0) return 0; + // Guard against Infinity from a huge exponent before the min() clamp. + const exp = Math.pow(multiplier, attempt); + const raw = Number.isFinite(exp) ? base * exp : Infinity; + return Math.min(cap, raw); +} + +/** + * Full-jitter delay for an attempt: a uniform random value in + * `[0, backoffCeiling(attempt)]`. `rng` is injectable for deterministic tests + * (defaults to `Math.random`). + */ +export function fullJitterBackoff( + attempt: number, + base: number = INITIAL_BACKOFF_MS, + cap: number = MAX_BACKOFF_MS, + rng: () => number = Math.random +): number { + const ceiling = backoffCeiling(attempt, base, cap); + // Clamp rng into [0, 1) defensively, then scale and round to whole ms. + const r = Math.min(0.999999, Math.max(0, rng())); + return Math.round(r * ceiling); +} diff --git a/src/utils/frameBuffer.ts b/src/utils/frameBuffer.ts new file mode 100644 index 0000000..e682f77 --- /dev/null +++ b/src/utils/frameBuffer.ts @@ -0,0 +1,93 @@ +/** + * Fixed-capacity circular buffer of telemetry frames used to replay recent data + * during the reconnection recovery handshake. Pushing past capacity overwrites + * the oldest frame (the operator only needs the most recent window to recover). + */ + +import { FRAME_BUFFER_CAPACITY, type TelemetryFrame } from "@/types/connection"; + +export class FrameBuffer { + private readonly buffer: (TelemetryFrame | undefined)[]; + private readonly capacity: number; + /** Index of the next write slot. */ + private head = 0; + /** Number of frames currently stored (≤ capacity). */ + private count = 0; + /** Highest sequence id seen, or -1 when empty. */ + private highestSeq = -1; + + constructor(capacity: number = FRAME_BUFFER_CAPACITY) { + if (capacity <= 0) throw new Error("FrameBuffer capacity must be positive"); + this.capacity = capacity; + this.buffer = new Array(capacity); + } + + /** Number of frames retained. */ + get size(): number { + return this.count; + } + + /** True once the ring has wrapped and is overwriting old frames. */ + get isFull(): boolean { + return this.count === this.capacity; + } + + /** Highest sequence id pushed, or -1 if the buffer is empty. */ + get lastSequenceId(): number { + return this.highestSeq; + } + + /** Append a frame, evicting the oldest if at capacity. */ + push(frame: TelemetryFrame): void { + this.buffer[this.head] = frame; + this.head = (this.head + 1) % this.capacity; + if (this.count < this.capacity) { + this.count += 1; + } + if (frame.sequenceId > this.highestSeq) { + this.highestSeq = frame.sequenceId; + } + } + + /** Return retained frames in chronological (oldest → newest) order. */ + toArray(): TelemetryFrame[] { + const out: TelemetryFrame[] = []; + // The oldest element is `count` slots behind head (mod capacity). + const start = (this.head - this.count + this.capacity) % this.capacity; + for (let i = 0; i < this.count; i++) { + const frame = this.buffer[(start + i) % this.capacity]; + if (frame) out.push(frame); + } + return out; + } + + /** + * Drain the buffer: return all retained frames (oldest → newest) and clear + * the ring. `highestSeq` is preserved so the next recovery handshake can still + * report the last sequence id the client observed. + */ + drain(): TelemetryFrame[] { + const frames = this.toArray(); + this.buffer.fill(undefined); + this.head = 0; + this.count = 0; + return frames; + } + + /** Approximate total bytes retained (sum of frame sizes). */ + byteLength(): number { + let total = 0; + for (const frame of this.buffer) { + if (frame) total += frame.size; + } + return total; + } + + /** Reset the buffer and forget the high-water sequence id. */ + clear(): void { + this.buffer.fill(undefined); + this.head = 0; + this.count = 0; + this.highestSeq = -1; + } +} diff --git a/tests/hooks/useWebSocket.test.ts b/tests/hooks/useWebSocket.test.ts new file mode 100644 index 0000000..791a7d9 --- /dev/null +++ b/tests/hooks/useWebSocket.test.ts @@ -0,0 +1,194 @@ +import { describe, it, expect, beforeEach, afterEach, vi } from "vitest"; +import { renderHook, act } from "@testing-library/react"; +import { useWebSocket } from "@/hooks/useWebSocket"; +import { connectionStore } from "@/store/slices/connectionSlice"; +import type { SocketLike } from "@/hooks/useTelemetryStream"; +import type { TelemetryFrame } from "@/types/connection"; + +class FakeSocket implements SocketLike { + static instances: FakeSocket[] = []; + binaryType = "arraybuffer"; + readyState = 0; + sent: string[] = []; + onopen: ((ev: Event) => void) | null = null; + onmessage: ((ev: MessageEvent) => void) | null = null; + onclose: ((ev: CloseEvent) => void) | null = null; + onerror: ((ev: Event) => void) | null = null; + + constructor(public url: string) { + FakeSocket.instances.push(this); + } + send(data: string | ArrayBuffer): void { + this.sent.push(typeof data === "string" ? data : ""); + } + close(): void { + this.readyState = 3; + this.onclose?.(new CloseEvent("close")); + } + fireOpen(): void { + this.readyState = 1; + this.onopen?.(new Event("open")); + } + fireMessage(payload: unknown): void { + const data = typeof payload === "string" ? payload : JSON.stringify(payload); + this.onmessage?.({ data } as MessageEvent); + } + lastSent(): unknown { + const raw = this.sent[this.sent.length - 1]; + try { + return JSON.parse(raw); + } catch { + return raw; + } + } +} + +function makeStorage() { + const map = new Map(); + return { + getItem: (k: string) => map.get(k) ?? null, + setItem: (k: string, v: string) => void map.set(k, v), + removeItem: (k: string) => void map.delete(k), + _map: map, + }; +} + +const baseOptions = () => ({ + url: "wss://api.test/ws", + createSocket: (url: string) => new FakeSocket(url), + rng: () => 0.5, + now: () => 1000, + storage: makeStorage(), +}); + +beforeEach(() => { + vi.useFakeTimers(); + FakeSocket.instances = []; + connectionStore.dispatch({ type: "RESET" }); +}); + +afterEach(() => { + vi.runOnlyPendingTimers(); + vi.useRealTimers(); +}); + +describe("useWebSocket", () => { + it("connects, opens, and reaches connected", () => { + const { result } = renderHook(() => useWebSocket(baseOptions())); + expect(FakeSocket.instances).toHaveLength(1); + + act(() => FakeSocket.instances[0].fireOpen()); + expect(result.current.status).toBe("connected"); + expect(connectionStore.getState().status).toBe("connected"); + }); + + it("answers server pings with a pong", () => { + renderHook(() => useWebSocket(baseOptions())); + const sock = FakeSocket.instances[0]; + act(() => sock.fireOpen()); + act(() => sock.fireMessage({ type: "ping" })); + expect(sock.lastSent()).toEqual({ type: "pong" }); + }); + + it("persists the assigned sticky node id and uses it on reconnect", () => { + const opts = baseOptions(); + renderHook(() => useWebSocket(opts)); + const sock = FakeSocket.instances[0]; + act(() => sock.fireOpen()); + act(() => sock.fireMessage({ type: "node_assigned", nodeId: "node-42" })); + + expect(opts.storage._map.get("ws:stickyNodeId")).toBe("node-42"); + + // Drop the connection and let the backoff timer fire. + act(() => sock.close()); + act(() => vi.advanceTimersByTime(1000)); + const reconnected = FakeSocket.instances[1]; + expect(reconnected.url).toContain("nodeId=node-42"); + }); + + it("buffers telemetry frames and forwards them to onFrame", () => { + const frames: TelemetryFrame[] = []; + const opts = { ...baseOptions(), onFrame: (f: TelemetryFrame) => frames.push(f) }; + renderHook(() => useWebSocket(opts)); + const sock = FakeSocket.instances[0]; + act(() => sock.fireOpen()); + act(() => sock.fireMessage({ sequenceId: 1, data: { x: 1 }, receivedAt: 1, size: 50 })); + act(() => sock.fireMessage({ sequenceId: 2, data: { x: 2 }, receivedAt: 2, size: 50 })); + expect(frames.map((f) => f.sequenceId)).toEqual([1, 2]); + }); + + it("schedules a jittered reconnect, runs recovery, and backfills missed frames", async () => { + const replayFetch = vi + .fn<(from: number, to: number) => Promise>() + .mockResolvedValue([ + { sequenceId: 6, data: {}, receivedAt: 6, size: 10 }, + ]); + const backfilled: number[] = []; + const opts = { + ...baseOptions(), + replayFetch, + onFrame: (f: TelemetryFrame) => backfilled.push(f.sequenceId), + }; + const { result } = renderHook(() => useWebSocket(opts)); + + const sock = FakeSocket.instances[0]; + act(() => sock.fireOpen()); + act(() => sock.fireMessage({ sequenceId: 5, data: {}, receivedAt: 5, size: 10 })); + + // Server drops us. + act(() => sock.close()); + expect(result.current.status).toBe("reconnecting"); + expect(result.current.attempt).toBe(1); + + // ceiling(0)=500, rng 0.5 → 250 ms backoff. + act(() => vi.advanceTimersByTime(250)); + expect(FakeSocket.instances).toHaveLength(2); + + const sock2 = FakeSocket.instances[1]; + act(() => sock2.fireOpen()); + // Reconnect enters recovery and replays the buffered frame. + expect(result.current.status).toBe("recovering"); + const recovery = sock2.lastSent() as { type: string; lastSequenceId: number }; + expect(recovery.type).toBe("recovery"); + expect(recovery.lastSequenceId).toBe(5); + + await act(async () => { + sock2.fireMessage({ + type: "recovery_ack", + missedCount: 1, + serverCurrentSeq: 6, + }); + await Promise.resolve(); + await Promise.resolve(); + }); + + expect(result.current.status).toBe("connected"); + expect(replayFetch).toHaveBeenCalledWith(5, 6); + expect(backfilled).toContain(6); + }); + + it("treats a missed heartbeat as a dead link and reconnects", () => { + const { result } = renderHook(() => useWebSocket(baseOptions())); + const sock = FakeSocket.instances[0]; + act(() => sock.fireOpen()); + expect(result.current.status).toBe("connected"); + + // No ping arrives within interval + timeout (20s) → watchdog fires. + act(() => vi.advanceTimersByTime(20_000)); + expect(result.current.missedHeartbeats).toBe(1); + expect(result.current.status).toBe("reconnecting"); + }); + + it("surfaces terminal failure after the max attempts", () => { + const { result } = renderHook(() => useWebSocket(baseOptions())); + // Fail every connection attempt by closing immediately on open-less sockets. + act(() => FakeSocket.instances[0].close()); + for (let i = 1; i <= 12; i++) { + act(() => vi.advanceTimersByTime(30_000)); + const latest = FakeSocket.instances[FakeSocket.instances.length - 1]; + act(() => latest.close()); + } + expect(result.current.status).toBe("failed"); + expect(connectionStore.getState().status).toBe("failed"); + }); +}); diff --git a/tests/unit/backoff.test.ts b/tests/unit/backoff.test.ts new file mode 100644 index 0000000..65e527e --- /dev/null +++ b/tests/unit/backoff.test.ts @@ -0,0 +1,58 @@ +import { describe, it, expect } from "vitest"; +import { backoffCeiling, fullJitterBackoff } from "@/utils/backoff"; + +describe("backoffCeiling", () => { + it("grows exponentially from the base", () => { + expect(backoffCeiling(0, 500, 30_000)).toBe(500); + expect(backoffCeiling(1, 500, 30_000)).toBe(1000); + expect(backoffCeiling(2, 500, 30_000)).toBe(2000); + expect(backoffCeiling(3, 500, 30_000)).toBe(4000); + }); + + it("clamps to the cap", () => { + // 500 * 2^7 = 64000 > 30000 cap + expect(backoffCeiling(7, 500, 30_000)).toBe(30_000); + expect(backoffCeiling(20, 500, 30_000)).toBe(30_000); + }); + + it("never overflows to Infinity for huge attempts", () => { + expect(backoffCeiling(2000, 500, 30_000)).toBe(30_000); + }); + + it("returns 0 for negative attempts", () => { + expect(backoffCeiling(-1)).toBe(0); + }); +}); + +describe("fullJitterBackoff", () => { + it("returns 0 when rng yields 0", () => { + expect(fullJitterBackoff(3, 500, 30_000, () => 0)).toBe(0); + }); + + it("returns ~half the ceiling at rng=0.5", () => { + // ceiling(3) = 4000 → 0.5 * 4000 = 2000 + expect(fullJitterBackoff(3, 500, 30_000, () => 0.5)).toBe(2000); + }); + + it("never exceeds the ceiling even at rng→1", () => { + const ceiling = backoffCeiling(2, 500, 30_000); // 2000 + const value = fullJitterBackoff(2, 500, 30_000, () => 0.9999999); + expect(value).toBeLessThanOrEqual(ceiling); + expect(value).toBeGreaterThan(0); + }); + + it("stays within [0, cap] across many random draws", () => { + let seed = 1; + const rng = () => { + seed = (seed * 1103515245 + 12345) & 0x7fffffff; + return seed / 0x7fffffff; + }; + for (let attempt = 0; attempt < 12; attempt++) { + for (let i = 0; i < 50; i++) { + const d = fullJitterBackoff(attempt, 500, 30_000, rng); + expect(d).toBeGreaterThanOrEqual(0); + expect(d).toBeLessThanOrEqual(30_000); + } + } + }); +}); diff --git a/tests/unit/connectionSlice.test.ts b/tests/unit/connectionSlice.test.ts new file mode 100644 index 0000000..4b2170c --- /dev/null +++ b/tests/unit/connectionSlice.test.ts @@ -0,0 +1,115 @@ +import { describe, it, expect, beforeEach } from "vitest"; +import { + connectionStore, + deriveQuality, +} from "@/store/slices/connectionSlice"; + +describe("deriveQuality", () => { + it("is good when connected, no missed beats, low latency", () => { + expect(deriveQuality("connected", 0, 20)).toBe("good"); + }); + + it("is degraded when connected but a heartbeat was missed", () => { + expect(deriveQuality("connected", 1, 20)).toBe("degraded"); + }); + + it("is degraded when connected but latency exceeds the heartbeat timeout", () => { + expect(deriveQuality("connected", 0, 6000)).toBe("degraded"); + }); + + it("is degraded for transitional states", () => { + expect(deriveQuality("connecting", 0, -1)).toBe("degraded"); + expect(deriveQuality("reconnecting", 0, -1)).toBe("degraded"); + expect(deriveQuality("recovering", 0, -1)).toBe("degraded"); + }); + + it("is down when failed or idle", () => { + expect(deriveQuality("failed", 0, -1)).toBe("down"); + expect(deriveQuality("idle", 0, -1)).toBe("down"); + }); +}); + +describe("connectionStore", () => { + beforeEach(() => connectionStore.dispatch({ type: "RESET" })); + + it("updates status and derives quality", () => { + connectionStore.dispatch({ + type: "CONNECTION_STATUS_CHANGED", + payload: { status: "connected" }, + }); + const s = connectionStore.getState(); + expect(s.status).toBe("connected"); + expect(s.quality).toBe("good"); + }); + + it("tracks reconnect attempts and scheduled delay", () => { + connectionStore.dispatch({ + type: "CONNECTION_STATUS_CHANGED", + payload: { status: "reconnecting", attempt: 3, nextDelayMs: 1234 }, + }); + const s = connectionStore.getState(); + expect(s.status).toBe("reconnecting"); + expect(s.attempt).toBe(3); + expect(s.nextDelayMs).toBe(1234); + expect(s.quality).toBe("degraded"); + }); + + it("clears missed heartbeats when (re)connected", () => { + connectionStore.dispatch({ + type: "HEARTBEAT_MISSED", + payload: { missedHeartbeats: 2 }, + }); + expect(connectionStore.getState().missedHeartbeats).toBe(2); + connectionStore.dispatch({ + type: "CONNECTION_STATUS_CHANGED", + payload: { status: "connected" }, + }); + expect(connectionStore.getState().missedHeartbeats).toBe(0); + }); + + it("stores the sticky node id", () => { + connectionStore.dispatch({ + type: "NODE_ID_ASSIGNED", + payload: { nodeId: "node-7" }, + }); + expect(connectionStore.getState().nodeId).toBe("node-7"); + }); + + it("records terminal failure with an error and down quality", () => { + connectionStore.dispatch({ + type: "CONNECTION_FAILED", + payload: { error: "gave up" }, + }); + const s = connectionStore.getState(); + expect(s.status).toBe("failed"); + expect(s.quality).toBe("down"); + expect(s.lastError).toBe("gave up"); + }); + + it("CONNECTION_RECOVERED resets attempts and clears the error", () => { + connectionStore.dispatch({ + type: "CONNECTION_STATUS_CHANGED", + payload: { status: "reconnecting", attempt: 5 }, + }); + connectionStore.dispatch({ type: "CONNECTION_RECOVERED" }); + const s = connectionStore.getState(); + expect(s.status).toBe("connected"); + expect(s.attempt).toBe(0); + expect(s.lastError).toBeNull(); + }); + + it("notifies subscribers", () => { + const seen: string[] = []; + const unsub = connectionStore.subscribe((s) => seen.push(s.status)); + connectionStore.dispatch({ + type: "CONNECTION_STATUS_CHANGED", + payload: { status: "connecting" }, + }); + unsub(); + connectionStore.dispatch({ + type: "CONNECTION_STATUS_CHANGED", + payload: { status: "connected" }, + }); + expect(seen).toEqual(["connecting"]); + }); +}); diff --git a/tests/unit/frameBuffer.test.ts b/tests/unit/frameBuffer.test.ts new file mode 100644 index 0000000..34ffcfb --- /dev/null +++ b/tests/unit/frameBuffer.test.ts @@ -0,0 +1,70 @@ +import { describe, it, expect } from "vitest"; +import { FrameBuffer } from "@/utils/frameBuffer"; +import type { TelemetryFrame } from "@/types/connection"; + +function frame(seq: number, size = 100): TelemetryFrame { + return { sequenceId: seq, data: { v: seq }, receivedAt: seq, size }; +} + +describe("FrameBuffer", () => { + it("retains frames in chronological order", () => { + const buf = new FrameBuffer(4); + buf.push(frame(1)); + buf.push(frame(2)); + buf.push(frame(3)); + expect(buf.size).toBe(3); + expect(buf.toArray().map((f) => f.sequenceId)).toEqual([1, 2, 3]); + }); + + it("evicts the oldest frames once full (ring behaviour)", () => { + const buf = new FrameBuffer(3); + [1, 2, 3, 4, 5].forEach((s) => buf.push(frame(s))); + expect(buf.isFull).toBe(true); + expect(buf.size).toBe(3); + expect(buf.toArray().map((f) => f.sequenceId)).toEqual([3, 4, 5]); + }); + + it("tracks the highest sequence id", () => { + const buf = new FrameBuffer(3); + buf.push(frame(10)); + buf.push(frame(11)); + expect(buf.lastSequenceId).toBe(11); + }); + + it("drains all frames and clears, preserving lastSequenceId", () => { + const buf = new FrameBuffer(5); + [1, 2, 3].forEach((s) => buf.push(frame(s))); + const drained = buf.drain(); + expect(drained.map((f) => f.sequenceId)).toEqual([1, 2, 3]); + expect(buf.size).toBe(0); + expect(buf.toArray()).toEqual([]); + // lastSequenceId survives a drain so the next recovery handshake is correct. + expect(buf.lastSequenceId).toBe(3); + }); + + it("sums byte sizes", () => { + const buf = new FrameBuffer(5); + buf.push(frame(1, 200)); + buf.push(frame(2, 300)); + expect(buf.byteLength()).toBe(500); + }); + + it("clear() resets the high-water mark", () => { + const buf = new FrameBuffer(5); + buf.push(frame(7)); + buf.clear(); + expect(buf.size).toBe(0); + expect(buf.lastSequenceId).toBe(-1); + }); + + it("rejects a non-positive capacity", () => { + expect(() => new FrameBuffer(0)).toThrow(); + }); + + it("defaults to a capacity of 500", () => { + const buf = new FrameBuffer(); + for (let i = 0; i < 600; i++) buf.push(frame(i)); + expect(buf.size).toBe(500); + expect(buf.toArray()[0].sequenceId).toBe(100); // oldest 100 evicted + }); +}); diff --git a/tests/unit/reconnectMachine.test.ts b/tests/unit/reconnectMachine.test.ts new file mode 100644 index 0000000..1407a50 --- /dev/null +++ b/tests/unit/reconnectMachine.test.ts @@ -0,0 +1,126 @@ +import { describe, it, expect } from "vitest"; +import { ReconnectMachine } from "@/services/reconnectMachine"; + +/** Deterministic machine: jitter rng pinned to 0.5. */ +function machine() { + return new ReconnectMachine({ rng: () => 0.5 }); +} + +describe("ReconnectMachine", () => { + it("connects from idle and reaches connected on first open", () => { + const m = machine(); + m.send({ type: "CONNECT" }); + expect(m.status).toBe("connecting"); + m.send({ type: "CONNECTED" }); + expect(m.status).toBe("connected"); + expect(m.getContext().attempt).toBe(0); + }); + + it("enters reconnecting with a jittered delay on disconnect", () => { + const m = machine(); + m.send({ type: "CONNECT" }); + m.send({ type: "CONNECTED" }); + m.send({ type: "DISCONNECTED" }); + const ctx = m.getContext(); + expect(ctx.status).toBe("reconnecting"); + expect(ctx.attempt).toBe(1); + // ceiling(0) = 500, rng 0.5 → 250 + expect(ctx.nextDelayMs).toBe(250); + }); + + it("RETRY moves reconnecting → connecting", () => { + const m = machine(); + m.send({ type: "CONNECT" }); + m.send({ type: "CONNECTED" }); + m.send({ type: "DISCONNECTED" }); + m.send({ type: "RETRY" }); + expect(m.status).toBe("connecting"); + expect(m.getContext().nextDelayMs).toBeNull(); + }); + + it("re-open after an outage routes through recovering, not connected", () => { + const m = machine(); + m.send({ type: "CONNECT" }); + m.send({ type: "CONNECTED" }); + m.send({ type: "DISCONNECTED" }); // attempt 1 + m.send({ type: "RETRY" }); + m.send({ type: "CONNECTED" }); + expect(m.status).toBe("recovering"); + m.send({ type: "RECOVERY_SUCCESS" }); + expect(m.status).toBe("connected"); + expect(m.getContext().attempt).toBe(0); + }); + + it("backoff ceiling grows with each consecutive failed attempt", () => { + const m = machine(); + m.send({ type: "CONNECT" }); + m.send({ type: "CONNECTED" }); + const delays: number[] = []; + for (let i = 0; i < 4; i++) { + m.send({ type: "DISCONNECTED" }); + delays.push(m.getContext().nextDelayMs!); + m.send({ type: "RETRY" }); + } + // rng 0.5 × ceiling(0..3) = 0.5 × [500,1000,2000,4000] + expect(delays).toEqual([250, 500, 1000, 2000]); + }); + + it("declares terminal failure after the max attempts", () => { + const m = new ReconnectMachine({ rng: () => 0.5, maxAttempts: 3 }); + m.send({ type: "CONNECT" }); + m.send({ type: "CONNECTED" }); + // 3 allowed attempts, the 4th tips into failed. + for (let i = 0; i < 3; i++) { + m.send({ type: "DISCONNECTED" }); + expect(m.status).toBe("reconnecting"); + m.send({ type: "RETRY" }); + } + m.send({ type: "DISCONNECTED" }); + expect(m.status).toBe("failed"); + expect(m.isTerminal).toBe(true); + expect(m.getContext().lastError).toMatch(/after 3 attempts/); + }); + + it("a failed machine can be manually reconnected", () => { + const m = new ReconnectMachine({ rng: () => 0.5, maxAttempts: 1 }); + m.send({ type: "CONNECT" }); + m.send({ type: "CONNECTED" }); + m.send({ type: "DISCONNECTED" }); // attempt 1 + m.send({ type: "RETRY" }); + m.send({ type: "DISCONNECTED" }); // attempt 2 > max → failed + expect(m.status).toBe("failed"); + m.send({ type: "CONNECT" }); + expect(m.status).toBe("connecting"); + expect(m.getContext().attempt).toBe(0); + }); + + it("heartbeat timeout triggers a reconnect", () => { + const m = machine(); + m.send({ type: "CONNECT" }); + m.send({ type: "CONNECTED" }); + const count = m.recordMissedHeartbeat(); + expect(count).toBe(1); + m.send({ type: "HEARTBEAT_TIMEOUT" }); + expect(m.status).toBe("reconnecting"); + expect(m.getContext().lastError).toMatch(/Heartbeat/); + }); + + it("RESET returns to idle from any state", () => { + const m = machine(); + m.send({ type: "CONNECT" }); + m.send({ type: "CONNECTED" }); + m.send({ type: "DISCONNECTED" }); + m.send({ type: "RESET" }); + expect(m.status).toBe("idle"); + expect(m.getContext().attempt).toBe(0); + }); + + it("notifies subscribers on transitions", () => { + const m = machine(); + const seen: string[] = []; + m.subscribe((ctx) => seen.push(ctx.status)); + m.send({ type: "CONNECT" }); + m.send({ type: "CONNECTED" }); + expect(seen).toEqual(["connecting", "connected"]); + }); +});