diff --git a/AGENTS.md b/AGENTS.md index 3d9a85e..6b79995 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -80,6 +80,10 @@ ws.send('listening.generate', {}, null, (p) => { - **No client-side timeout.** If the backend never emits `:response`/`:error`/`:timeout`, the Promise **hangs forever**. Make sure the handler always replies (return a value, or `$this->success/error`, or throw). +- **Connection lost mid-request.** A send already written to a socket that then closes stays + pending by default. Set `rejectOnDisconnect: true` (≥ 0.3.2) to have it reject with + `{ message, code: 'connection_lost' }` instead. Sends still queued while disconnected are + unaffected — they go out after the reconnect. - **The reject value is `msg.data`**, not an `Error`. Read `e?.message || e?.error`. - **`setAppReady()` gate** — see §3. Forget it and *every* non-protocol `send()` hangs. - `data` must be an object. Arrays in params get comma-joined by `serializeParams`; encode @@ -150,13 +154,19 @@ ws.destroy() // full teardown > internal `_appReadyPromise` **forever**, and auto-reconnect-on-close won't fire. > `createFromNuxtConfig()` calls it for you. A hand-rolled `createVueWsClient` singleton **must** > call it once at startup, **after** the bearer/auth is set but **before** any feature `send()`. -> (LATC does this in `app.vue` after `Promise.race([... ws.connect()], 4s cap)`.) +> Call it **before** (not after awaiting) the first `ws.connect()`: it also arms +> auto-reconnect, so a failed first attempt is retried. Before 0.3.2, sends queued behind a +> failed first connect hung forever (blank page until reload); 0.3.2 resolves them on the +> next successful connection. Feature code should **not** call `connect()`/`setAppReady()` — that's the app bootstrap's job, exactly once. Reconnect/heartbeat are built in (don't reinvent): `connect()` throttles via `reconnectThrottle` -(3000ms — bypass for an immediate reconnect by setting `ws.last_reconnect_try = 0`), heartbeat ping +(3000ms — bypass for an immediate reconnect by setting `ws.last_reconnect_try = 0`; since 0.3.2 a +throttled call is retried when the window opens instead of being dropped), heartbeat ping every `heartbeatInterval` (20000ms), auto-reconnect after `reconnectDelay` on close when app-ready. +`connect(true)` replaces the socket; the old socket's late close no longer clobbers the new one, +and `destroy()` cancels a pending auto-reconnect (both fixed in 0.3.2). --- @@ -215,6 +225,7 @@ const ws = createWsClient({ url, getAuthToken: () => api.bearerToken }) // ✅ - [ ] Use the app's exported `ws` singleton — never call a factory twice. - [ ] `setAppReady()` must run once at startup or every `send()` hangs. - [ ] `send()` has **no timeout** — a non-replying backend hangs the Promise. +- [ ] A send in flight when the socket drops stays pending unless `rejectOnDisconnect: true`. - [ ] `send()` resolves the **unwrapped** payload (not `{ data }`); rejects with `msg.data` (not an Error). - [ ] Streaming = the 4th `progress` arg of `send()`. Don't hand-wire `:progress` listeners. - [ ] In components prefer `listenWhileMounted` (auto-teardown + channel filter). diff --git a/README.md b/README.md index 1a41883..df294aa 100644 --- a/README.md +++ b/README.md @@ -171,6 +171,7 @@ const api = createApiClient({ | `heartbeatInterval` | `number` | `20000` | Ping interval (ms) | | `reconnectDelay` | `number` | `3000` | Delay before reconnect attempt (ms) | | `autoReconnect` | `boolean` | `true` | Auto-reconnect on disconnect | +| `rejectOnDisconnect` | `boolean` | `false` | Reject in-flight sends with `{ code: 'connection_lost' }` when their socket closes (≥ 0.3.2) | | `isServer` | `boolean \| () => boolean` | `false` | Returns safe no-op stub when true | | `isNativePlatform` | `boolean \| () => boolean` | `false` | Suppresses browser-only notifications | diff --git a/package-lock.json b/package-lock.json index 6664b78..a838662 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "@blax-software/networking", - "version": "0.3.0", + "version": "0.3.2", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "@blax-software/networking", - "version": "0.3.0", + "version": "0.3.2", "license": "MIT", "devDependencies": { "axios": "^1.7.0", diff --git a/package.json b/package.json index c50e84c..46bf2d3 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@blax-software/networking", - "version": "0.3.1", + "version": "0.3.2", "description": "Plug-and-play API + WebSocket client. Framework-agnostic core with optional Vue, Nuxt, and React bindings.", "type": "module", "main": "./dist/index.cjs", @@ -37,7 +37,8 @@ "dev": "tsup --watch", "typecheck": "tsc --noEmit", "prepare": "tsup", - "prepublishOnly": "npm run build" + "prepublishOnly": "npm run build", + "test": "tsup && node --test test/*.test.mjs" }, "peerDependencies": { "axios": ">=1.0.0", diff --git a/src/types.ts b/src/types.ts index 1e9c0dc..fb0816d 100644 --- a/src/types.ts +++ b/src/types.ts @@ -161,6 +161,15 @@ export interface WsClientConfig { /** Auto-reconnect on connection loss. Default: `true` */ autoReconnect?: boolean + /** + * Reject a `send()` whose socket closes before the server answered, with + * `{ message, code: 'connection_lost' }`, instead of leaving the Promise + * pending forever. Sends still waiting to be written (queued while + * disconnected) are unaffected — they go out after the reconnect. + * Default: `false` (pre-0.3.2 behaviour). + */ + rejectOnDisconnect?: boolean + /** * Show connection lost/restored notifications. * Set to `false` on native mobile apps where a toast overlay may be unwanted. diff --git a/src/ws.ts b/src/ws.ts index 2c2f89f..4b640b1 100644 --- a/src/ws.ts +++ b/src/ws.ts @@ -15,6 +15,13 @@ interface WsSocket extends WebSocket { socket_id?: string } +/** A send written to a socket, tracked for `rejectOnDisconnect`. */ +interface InflightSend { + socket: WsSocket | null + fail: ((err: { message: string; code: string }) => void) | null + lost: { message: string; code: string } | null +} + export interface WsChannel { name: string is_established: boolean @@ -195,6 +202,16 @@ class WsClientImpl extends EventTarget implements WsClient { // Connect promise coalescing private _connectPromise: Promise | null = null + // A connect() that landed on the reconnect throttle is retried when the + // throttle window opens (instead of being dropped). + private _deferredConnectTimer: ReturnType | null = null + // Auto-reconnect scheduled by a close; destroy() cancels it. + private _reconnectTimer: ReturnType | null = null + + // rejectOnDisconnect: sends written to a socket, keyed by that socket. + private _inflight = new Set() + private _payloadSend = new WeakMap() + constructor(config: WsClientConfig, createRef: CreateRefFn) { super() this._config = { @@ -266,12 +283,13 @@ class WsClientImpl extends EventTarget implements WsClient { ensureConnected(): Promise { if (this.socket?.socket_id && this.is_opened.value) return Promise.resolve() - if (!this._connectedPromise) { - this._connectedPromise = new Promise((r) => { - this._connectedResolve = r - }) - this.connect().catch(() => {}) - } + // One waiter per outage. It is resolved by whichever connection attempt + // succeeds next — a failed attempt must not orphan it (that left every + // send queued behind a failed first connect hanging forever). + this._connectedPromise ??= new Promise((r) => { + this._connectedResolve = r + }) + this.connect().catch(() => {}) return this._connectedPromise } @@ -302,8 +320,10 @@ class WsClientImpl extends EventTarget implements WsClient { if (this._connectPromise) return this._connectPromise const throttle = this._config.reconnectThrottle ?? 3000 - if (this.last_reconnect_try && Date.now() - this.last_reconnect_try < throttle) { - console.log('[ws] Reconnect too fast, skipping') + const sinceLast = Date.now() - this.last_reconnect_try + if (this.last_reconnect_try && sinceLast < throttle) { + console.log('[ws] Reconnect too fast, deferring') + this._deferConnect(throttle - sinceLast) return } @@ -336,20 +356,33 @@ class WsClientImpl extends EventTarget implements WsClient { const socket = this.socket! socket.addEventListener('error', () => { - this.is_connecting_socket.value = false - this._connectPromise = null + // A replaced (connect(true)) or destroyed socket must not touch the + // live connection's state. + if (this.socket === socket) { + this.is_connecting_socket.value = false + this._connectPromise = null + } reject(new Error('WebSocket error')) }) socket.addEventListener('close', () => { + this._failInflight(socket) + + // Stale socket (replaced by connect(true), or destroy()ed): its late + // close must not wipe the live socket or schedule a reconnect. + if (this.socket !== socket) { + reject(new Error('Socket closed')) + return + } + this.channels = [] this.is_connecting_socket.value = false this.is_opened.value = false this.is_setup.value = false this.socket = null this._connectPromise = null - this._connectedPromise = null - this._connectedResolve = null + // Keep _connectedPromise / _connectedResolve: callers waiting in + // ensureConnected() are resolved by the next successful connection. this.is_after_lost_connection.value = true if (this._shouldNotify()) { @@ -362,7 +395,11 @@ class WsClientImpl extends EventTarget implements WsClient { if (this._appReady && this._config.autoReconnect !== false) { const delay = this._config.reconnectDelay ?? 3000 this._config.onConnectionStateChange?.('reconnecting') - setTimeout(() => this.connect().catch(() => {}), delay) + if (this._reconnectTimer) clearTimeout(this._reconnectTimer) + this._reconnectTimer = setTimeout(() => { + this._reconnectTimer = null + this.connect().catch(() => {}) + }, delay) } reject(new Error('Socket closed')) @@ -400,11 +437,14 @@ class WsClientImpl extends EventTarget implements WsClient { if (msg?.event === 'websocket.connection_established') { const data = JSON.parse(msg.data) - if (data?.socket_id && this.socket) { + if (data?.socket_id && this.socket === socket) { this.socket.socket_id = data.socket_id this.is_connecting_socket.value = false resolve(this.socket) - this._connectedResolve?.() + const waiting = this._connectedResolve + this._connectedPromise = null + this._connectedResolve = null + waiting?.() // Proactively establish the default channel this.channel() this._workSendQueue() @@ -445,6 +485,8 @@ class WsClientImpl extends EventTarget implements WsClient { for (const payload of queue) { await this.channel(payload.channel) this.socket?.send(JSON.stringify(payload)) + const tracked = this._payloadSend.get(payload) + if (tracked) tracked.socket = this.socket } } @@ -478,20 +520,37 @@ class WsClientImpl extends EventTarget implements WsClient { const payload = { event: sendingevent, data, channel: channel_name } - // Ensure the target channel is subscribed - if (channel_name && !_isProtocolEvent(event)) { - await this.channel(channel_name) + // rejectOnDisconnect: remember which socket this send went out on, so a + // close of that socket rejects it instead of leaving it pending forever. + const tracked: InflightSend | null = + this._config.rejectOnDisconnect === true && !_isProtocolEvent(event) + ? { socket: null, fail: null, lost: null } + : null + if (tracked) { + this._inflight.add(tracked) + this._payloadSend.set(payload, tracked) } - // Send or queue - if (this.is_opened.value && this.socket?.readyState === WebSocket.OPEN) { - this.socket.send(JSON.stringify(payload)) - } else { - this.send_queue.push(payload) - } + try { + // Ensure the target channel is subscribed + if (channel_name && !_isProtocolEvent(event)) { + await this.channel(channel_name) + } - await this.connect() - if (!this.socket) throw new Error('Socket not connected') + // Send or queue + if (this.is_opened.value && this.socket?.readyState === WebSocket.OPEN) { + this.socket.send(JSON.stringify(payload)) + if (tracked) tracked.socket = this.socket + } else { + this.send_queue.push(payload) + } + + await this.connect() + if (!this.socket) throw new Error('Socket not connected') + } catch (error) { + if (tracked) this._inflight.delete(tracked) + throw error + } const startTime = typeof performance !== 'undefined' ? performance.now() : Date.now() @@ -501,6 +560,7 @@ class WsClientImpl extends EventTarget implements WsClient { this.removeEventListener(sendingevent + ':progress', handler) this.removeEventListener(sendingevent + ':error', handler) this.removeEventListener(sendingevent + ':response', handler) + if (tracked) this._inflight.delete(tracked) } const handler = (m: any) => { @@ -542,6 +602,15 @@ class WsClientImpl extends EventTarget implements WsClient { this.addEventListener(sendingevent + ':progress', handler) this.addEventListener(sendingevent + ':error', handler) this.addEventListener(sendingevent + ':response', handler) + + if (tracked) { + tracked.fail = (err) => { + cleanup() + reject(err) + } + // The socket already closed between writing and waiting. + if (tracked.lost) tracked.fail(tracked.lost) + } }).catch((error: any) => { // If the server dropped our subscription, re-establish and retry once if ( @@ -562,6 +631,29 @@ class WsClientImpl extends EventTarget implements WsClient { }) as Promise } + /** Retry a throttled connect() once the throttle window has passed. */ + private _deferConnect(ms: number): void { + if (this._deferredConnectTimer) return + this._deferredConnectTimer = setTimeout(() => { + this._deferredConnectTimer = null + if (!this.socket && !this._connectPromise) this.connect().catch(() => {}) + }, Math.max(0, ms) + 5) + } + + /** rejectOnDisconnect: reject every send that went out on `socket`. */ + private _failInflight(socket: WsSocket): void { + for (const entry of [...this._inflight]) { + if (entry.socket !== socket) continue + this._inflight.delete(entry) + const err = { + message: this._t('websocket.requestlost', 'The connection was lost before the server answered.'), + code: 'connection_lost', + } + if (entry.fail) entry.fail(err) + else entry.lost = err + } + } + // ------------------------------------------------------------------------- // Event listeners (framework-agnostic — return cleanup function) // ------------------------------------------------------------------------- @@ -607,6 +699,10 @@ class WsClientImpl extends EventTarget implements WsClient { destroy(): void { if (this.heartbeat) clearInterval(this.heartbeat) this.heartbeat = null + if (this._deferredConnectTimer) clearTimeout(this._deferredConnectTimer) + this._deferredConnectTimer = null + if (this._reconnectTimer) clearTimeout(this._reconnectTimer) + this._reconnectTimer = null if (this.socket) { try { this.socket.close() diff --git a/test/ws.test.mjs b/test/ws.test.mjs new file mode 100644 index 0000000..19c8cae --- /dev/null +++ b/test/ws.test.mjs @@ -0,0 +1,214 @@ +// Connection-loss behaviour of the WS client, against a scripted fake socket. +// Run: npm test (builds dist/ first, then `node --test test/`). +import { test, beforeEach, afterEach } from 'node:test' +import assert from 'node:assert/strict' +import { createWsClient } from '../dist/index.js' + +// ── Fake WebSocket + tiny scripted server ────────────────────────────────── +class FakeWS extends EventTarget { + static CONNECTING = 0 + static OPEN = 1 + static CLOSING = 2 + static CLOSED = 3 + /** Per connection attempt: 'accept' | 'fail' | 'hang' (opens, never established). */ + static plan = [] + static sockets = [] + /** event name → false to never answer it. */ + static answer = {} + + constructor(url) { + super() + this.url = url + this.readyState = FakeWS.CONNECTING + this.sent = [] + this.index = FakeWS.sockets.length + FakeWS.sockets.push(this) + const mode = FakeWS.plan[this.index] ?? 'accept' + setTimeout(() => { + if (this.readyState === FakeWS.CLOSED) return + if (mode === 'fail') return this.drop() + this.readyState = FakeWS.OPEN + this.dispatchEvent(new Event('open')) + if (mode === 'accept') this.push({ event: 'websocket.connection_established', data: JSON.stringify({ socket_id: `${this.index}.1` }) }) + }, 5) + } + + send(raw) { + const msg = JSON.parse(raw) + this.sent.push(msg) + const base = String(msg.event).replace(/\[.*\]$/, '') + if (base === 'websocket.ping' || FakeWS.answer[base] === false) return + if (FakeWS.answer[base] && FakeWS.answer[base] !== true && FakeWS.answer[base] !== this.index) return + setTimeout(() => { + if (this.readyState !== FakeWS.OPEN) return + if (msg.event === 'websocket.subscribe') { + this.push({ event: 'websocket.subscribe:response', data: { channel: msg.data.channel }, channel: msg.data.channel }) + } else { + this.push({ event: msg.event + ':response', data: { ok: true, via: this.index }, channel: msg.channel }) + } + }, 2) + } + + close() { + if (this.readyState === FakeWS.CLOSED) return + this.readyState = FakeWS.CLOSED + setTimeout(() => this.dispatchEvent(new Event('close')), 3) + } + + /** Network failure: error + close. */ + drop() { + this.readyState = FakeWS.CLOSED + this.dispatchEvent(new Event('error')) + this.dispatchEvent(new Event('close')) + } + + push(obj) { + this.dispatchEvent(new MessageEvent('message', { data: JSON.stringify(obj) })) + } +} + +const realWebSocket = globalThis.WebSocket +const clients = [] + +beforeEach(() => { + FakeWS.plan = [] + FakeWS.sockets = [] + FakeWS.answer = {} + globalThis.WebSocket = FakeWS +}) + +afterEach(() => { + for (const c of clients.splice(0)) c.destroy() + globalThis.WebSocket = realWebSocket +}) + +const client = (extra = {}) => { + const c = createWsClient({ + url: 'ws://test/app/key', + reconnectDelay: 20, + reconnectThrottle: 40, + heartbeatInterval: 60_000, + showConnectionNotifications: false, + ...extra, + }) + clients.push(c) + return c +} + +const sleep = (ms) => new Promise((r) => setTimeout(r, ms)) + +/** Resolves to the value, or the string 'PENDING' if still unsettled after ms. */ +const within = (p, ms = 800) => Promise.race([p, sleep(ms).then(() => 'PENDING')]) + +// ── Tests ─────────────────────────────────────────────────────────────────── + +test('baseline: a send resolves on a healthy connection', async () => { + const ws = client() + ws.setAppReady() + const r = await within(ws.send('dash.index')) + assert.deepEqual(r, { ok: true, via: 0 }) +}) + +test('a send waiting on the app\'s first connect survives that attempt failing (setAppReady first)', async () => { + // The app bootstraps the socket; pages send while it is still connecting. + FakeWS.plan = ['fail', 'accept'] + const ws = client() + ws.setAppReady() + ws.connect().catch(() => {}) + const r = await within(ws.send('dash.index')) + assert.deepEqual(r, { ok: true, via: 1 }, 'the send must not hang on the dropped connection promise') + assert.equal(ws.is_setup.value, true) +}) + +test('a send after a failed connect that landed on the reconnect throttle still completes (connect before setAppReady)', async () => { + FakeWS.plan = ['fail', 'accept'] + const ws = client() + await ws.connect().catch(() => {}) // fails fast; no auto-reconnect yet (app not ready) + ws.setAppReady() + const r = await within(ws.send('dash.index')) + assert.deepEqual(r, { ok: true, via: 1 }, 'a throttled connect must be retried, not dropped') +}) + +test('unchanged: a send that itself opens the connection rejects when that attempt fails', async () => { + FakeWS.plan = ['fail', 'accept'] + const ws = client() + ws.setAppReady() + const r = await within(ws.send('dash.index').then(() => 'RESOLVED', (e) => e?.message)) + assert.equal(r, 'WebSocket error') +}) + +test('a forced reconnect is not undone by the old socket\'s late close event', async () => { + const ws = client() + ws.setAppReady() + await within(ws.send('dash.index')) + ws.last_reconnect_try = 0 + await ws.connect(true) + await sleep(30) // the old socket's close event fires meanwhile + assert.equal(ws.socket, FakeWS.sockets[1], 'the live socket must survive the stale close') + const r = await within(ws.send('dash.index')) + assert.deepEqual(r, { ok: true, via: 1 }) +}) + +test('destroy() stays destroyed: the closing socket does not trigger an auto-reconnect', async () => { + const ws = client() + ws.setAppReady() + await within(ws.send('dash.index')) + ws.destroy() + await sleep(80) + assert.equal(FakeWS.sockets.length, 1, 'no new socket after destroy()') +}) + +test('a subscribe lost with its socket is retried on the next connection', async () => { + FakeWS.plan = ['accept', 'accept'] + FakeWS.answer = { 'websocket.subscribe': 1 } // only socket #1 answers subscribes + const ws = client() + ws.setAppReady() + const p = ws.send('dash.index') + await sleep(30) // subscribe sent on socket #0, unanswered + FakeWS.sockets[0].drop() + const r = await within(p) + assert.deepEqual(r, { ok: true, via: 1 }) +}) + +test('rejectOnDisconnect: a send in flight on a dropped socket rejects with connection_lost', async () => { + FakeWS.answer = { 'slow.call': false } + const ws = client({ rejectOnDisconnect: true }) + ws.setAppReady() + await within(ws.send('dash.index')) + const p = ws.send('slow.call') + await sleep(20) + FakeWS.sockets[0].drop() + const r = await within(p.then(() => 'RESOLVED', (e) => e)) + assert.equal(r?.code, 'connection_lost') + assert.ok(typeof r?.message === 'string' && r.message.length > 0) +}) + +test('default (no rejectOnDisconnect): an in-flight send stays pending — behaviour unchanged', async () => { + FakeWS.answer = { 'slow.call': false } + const ws = client() + ws.setAppReady() + await within(ws.send('dash.index')) + const p = ws.send('slow.call') + await sleep(20) + FakeWS.sockets[0].drop() + assert.equal(await within(p, 150), 'PENDING') +}) + +test('sends queued while disconnected are delivered after the reconnect', async () => { + const ws = client() + ws.setAppReady() + await within(ws.send('dash.index')) + FakeWS.sockets[0].drop() + const r = await within(ws.send('after.drop')) + assert.deepEqual(r, { ok: true, via: 1 }) +}) + +test('destroy() cancels an auto-reconnect that a drop had already scheduled', async () => { + const ws = client() + ws.setAppReady() + await within(ws.send('dash.index')) + FakeWS.sockets[0].drop() // schedules the auto-reconnect (reconnectDelay 20 ms) + ws.destroy() + await sleep(80) + assert.equal(FakeWS.sockets.length, 1, 'no reconnect after destroy()') +})