diff --git a/common/deno.json b/common/deno.json index a219e95..7ccfcac 100644 --- a/common/deno.json +++ b/common/deno.json @@ -1,6 +1,6 @@ { "name": "@atp/common", - "version": "0.1.0-alpha.8", + "version": "0.1.0-alpha.9", "exports": { ".": "./mod.ts", "./server": "./server.ts" diff --git a/common/util.ts b/common/util.ts index 813eece..6cadafa 100644 --- a/common/util.ts +++ b/common/util.ts @@ -180,10 +180,4 @@ export const mapDefined = ( } } return output; -}; - -export const isErrnoException = ( - err: unknown, -): err is NodeJS.ErrnoException => { - return !!err && (err as Record)["code"] !== undefined; -}; +}; \ No newline at end of file diff --git a/xrpc-server/deno.json b/xrpc-server/deno.json index a7b2523..9608fde 100644 --- a/xrpc-server/deno.json +++ b/xrpc-server/deno.json @@ -1,6 +1,6 @@ { "name": "@atp/xrpc-server", - "version": "0.1.0-alpha.7", + "version": "0.1.0-alpha.8", "exports": "./mod.ts", "license": "MIT", "imports": { diff --git a/xrpc-server/stream/websocket-keepalive.ts b/xrpc-server/stream/websocket-keepalive.ts index 0c927f6..9be3ce8 100644 --- a/xrpc-server/stream/websocket-keepalive.ts +++ b/xrpc-server/stream/websocket-keepalive.ts @@ -1,17 +1,41 @@ -import { type ClientOptions, createWebSocketStream, WebSocket } from "ws"; -import { isErrnoException, SECOND, wait } from "@atp/common"; +// websocket-keepalive.ts +// Runtime-agnostic (Deno / Workers / Bun / Browser) + +import { SECOND, wait } from "@atp/common"; import { CloseCode, DisconnectError } from "./types.ts"; +import { iterateBinary } from "./stream.ts"; -export type KeepAliveOptions = ClientOptions & { +/** + * Options for a {@link WebSocketKeepAlive} instance + * + * @prop getUrl Method to get the current URL of the websocket endpoint + * @prop maxReconnectSeconds Maximum time a request can take to reconnect + * @prop signal Abort signal to send when aborting connection + * + * @prop heartbeatIntervalMs Interval to send provided heartbeatPayload on, + * @prop heartbeatPayload Method to create payload to send for heartbeat + * @prop isPong If provided, we mark alive only when it returns true for a message + * if omitted, *any* message is considered proof of life + * + * @prop onReconnectError Reconnect hook + * + * @prop createSocket Socket factory override (lets you use custom client if needed) + * @prop protocols Override value for accepted protocols + */ +export type KeepAliveOptions = { getUrl: () => Promise; maxReconnectSeconds?: number; signal?: AbortSignal; - heartbeatIntervalMs?: number; - onReconnectError?: ( - error: unknown, - n: number, - initialSetup: boolean, - ) => void; + + heartbeatIntervalMs?: number; // default 10 * SECOND + heartbeatPayload?: () => string | ArrayBuffer | Uint8Array | Blob; + isPong?: (data: unknown) => boolean; + + // Reconnect hook + onReconnectError?: (error: unknown, n: number, initialSetup: boolean) => void; + + createSocket?: (url: string, protocols?: string | string[]) => WebSocket; + protocols?: string | string[]; }; export class WebSocketKeepAlive { @@ -19,10 +43,15 @@ export class WebSocketKeepAlive { public initialSetup = true; public reconnects: number | null = null; + /** + * Creates a new WebSocketKeepAlive instance. + * @param opts Configuration options for keepalive, heartbeat, reconnect, and socket creation. + */ constructor(public opts: KeepAliveOptions) {} async *[Symbol.asyncIterator](): AsyncGenerator { const maxReconnectMs = 1000 * (this.opts.maxReconnectSeconds ?? 64); + while (true) { if (this.reconnects !== null) { const duration = this.initialSetup @@ -30,85 +59,141 @@ export class WebSocketKeepAlive { : backoffMs(this.reconnects++, maxReconnectMs); await wait(duration); } + const url = await this.opts.getUrl(); - this.ws = new WebSocket(url, this.opts); + + // Create a web-standard WebSocket (or a custom one if provided). + const ws = this.opts.createSocket?.(url, this.opts.protocols) ?? + new WebSocket(url, this.opts.protocols); + this.ws = ws; + const ac = new AbortController(); if (this.opts.signal) { forwardSignal(this.opts.signal, ac); } - this.ws.once("open", () => { - this.initialSetup = false; - this.reconnects = 0; - if (this.ws) { - this.startHeartbeat(this.ws); - } - }); - this.ws.once("close", (code: number, reason: Uint8Array) => { - if (code === CloseCode.Abnormal) { - ac.abort( - new AbnormalCloseError( - `Abnormal ws close: ${new TextDecoder().decode(reason)}`, - ), - ); - } - }); + + // Track liveness (application-level heartbeat) + this.startHeartbeat(ws, ac); + + // When the socket opens, reset backoff. + ws.addEventListener( + "open", + () => { + this.initialSetup = false; + this.reconnects = 0; + }, + { once: true }, + ); + + // Distinguish abnormal close → treat as reconnectable error + ws.addEventListener( + "close", + (ev) => { + if (ev.code === CloseCode.Abnormal) { + ac.abort( + new AbnormalCloseError( + `Abnormal ws close: ${String(ev.reason || "")}`, + ), + ); + } + }, + { once: true }, + ); try { - const wsStream = createWebSocketStream(this.ws, { - signal: ac.signal, - readableObjectMode: true, - }); - for await (const chunk of wsStream) { + // Iterate incoming binary chunks + for await (const chunk of iterateBinary(ws)) { yield chunk; } - } catch (_err) { - const err = isErrnoException(_err) && _err.code === "ABORT_ERR" - ? _err.cause - : _err; + } catch (error) { + // Normalize Abort into same shape your old code expected. + const err = (error as Error)?.name === "AbortError" + ? (error as Error).cause ?? error + : error; + if (err instanceof DisconnectError) { - this.ws?.close(err.wsCode); + // We cleanly end the connection + ws?.close(err.wsCode); break; } - this.ws?.close(); + + // Close if not already closing + ws.close(); + if (isReconnectable(err)) { - this.reconnects ??= 0; + this.reconnects ??= 0; // Never reconnect when null this.opts.onReconnectError?.(err, this.reconnects, this.initialSetup); - continue; + continue; // loop to reconnect } else { throw err; } } + + // Other side ended stream cleanly; stop iterating. break; } } - startHeartbeat(ws: WebSocket) { + /** Application-level heartbeat (web standard). + * + * In Node's `ws` you used `ping`/`pong`. Those do not exist in web sockets. + * Here we: + * - periodically send `heartbeatPayload()` if provided + * - consider the connection "alive" when: + * * `isPong(ev.data)` returns true (if provided), OR + * * *any* message is received (fallback) + * - if no proof of life for one interval, we close the socket (which triggers reconnect) + */ + private startHeartbeat(ws: WebSocket, ac: AbortController) { + const intervalMs = this.opts.heartbeatIntervalMs ?? 10 * SECOND; + let isAlive = true; - let heartbeatInterval: ReturnType | null = null; + let timer: number | null = null; - const checkAlive = () => { + const onMessage = (ev: MessageEvent) => { + // If a custom pong detector exists, use it; otherwise any message counts. + if (!this.opts.isPong || this.opts.isPong(ev.data)) { + isAlive = true; + } + }; + + const tick = () => { if (!isAlive) { - return ws.terminate(); + // No pong/traffic since last tick → consider dead and close. + ws.close(1000); + // Abort the iterator with a recognizable shape like before. + const domErr = new DOMException("Aborted", "AbortError"); + domErr.cause = new DisconnectError( + CloseCode.Abnormal, + "HeartbeatTimeout", + ); + ac.abort(domErr); + return; } isAlive = false; - ws.ping(); + + const payload = this.opts.heartbeatPayload?.(); + if (payload !== undefined) { + ws.send(payload); + } }; - checkAlive(); - heartbeatInterval = setInterval( - checkAlive, - this.opts.heartbeatIntervalMs ?? 10 * SECOND, - ); + // Prime one cycle and schedule subsequent ones + tick(); + timer = setInterval(tick, intervalMs) as unknown as number; - ws.on("pong", () => { - isAlive = true; - }); - ws.once("close", () => { - if (heartbeatInterval) { - clearInterval(heartbeatInterval); - heartbeatInterval = null; - } - }); + ws.addEventListener("message", onMessage); + ws.addEventListener( + "close", + () => { + if (timer !== null) { + clearInterval(timer); + timer = null; + } + ws.removeEventListener("message", onMessage); + }, + { once: true }, + ); } } @@ -119,10 +204,11 @@ class AbnormalCloseError extends Error { } function isReconnectable(err: unknown): boolean { - if (isErrnoException(err) && typeof err.code === "string") { - return networkErrorCodes.includes(err.code); - } - return false; + // Network-ish errors are reconnectable. Keep your previous codes. + if (!err || typeof err !== "object") return false; + const e = err as { name?: unknown; code?: unknown }; + if (typeof e.name !== "string") return false; + return typeof e.code === "string" && networkErrorCodes.includes(e.code); } const networkErrorCodes = [ @@ -133,11 +219,12 @@ const networkErrorCodes = [ "EPIPE", "ETIMEDOUT", "ECANCELED", + "ABORT_ERR", // surface our aborts as reconnectable if you want ]; function backoffMs(n: number, maxMs: number) { - const baseSec = Math.pow(2, n); - const randSec = Math.random() - 0.5; + const baseSec = Math.pow(2, n); // 1, 2, 4, ... + const randSec = Math.random() - 0.5; // jitter [-0.5, +0.5] const ms = 1000 * (baseSec + randSec); return Math.min(ms, maxMs); } @@ -145,9 +232,8 @@ function backoffMs(n: number, maxMs: number) { function forwardSignal(signal: AbortSignal, ac: AbortController) { if (signal.aborted) { return ac.abort(signal.reason); - } else { - signal.addEventListener("abort", () => ac.abort(signal.reason), { - signal: ac.signal, - }); } -} + const onAbort = () => ac.abort(signal.reason); + // Use AbortSignal.any? Not universally available; just add/remove. + signal.addEventListener("abort", onAbort, { signal: ac.signal }); +} \ No newline at end of file