fix(ws): don't strand sends when a connection attempt fails (0.3.2)

A page's send() queued behind a failing first connect hung forever — the
close handler dropped the ensureConnected() waiter without settling it, and
a throttled connect() was silently skipped, so nothing ever reconnected.
Symptom in apps: sidebar renders, page body stays blank until a reload.

- ensureConnected(): one waiter per outage, resolved by whichever attempt
  succeeds next (no longer orphaned by a failed attempt)
- connect(): a call that lands on the reconnect throttle is retried when the
  window opens instead of being dropped
- a replaced (connect(true)) or destroy()ed socket's late error/close no
  longer wipes the live socket's state or schedules a reconnect
- destroy() cancels a pending auto-reconnect
- new opt-in `rejectOnDisconnect`: a send already written to a socket that
  closes rejects with { message, code: 'connection_lost' } instead of
  hanging (default off — unchanged behaviour for existing consumers)
- node:test suite (`npm test`) with a scripted fake socket covering each case

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Fabian @ Blax Software 2026-09-24 09:46:52 +02:00
parent c21553536f
commit 737954ba2d
7 changed files with 364 additions and 32 deletions

View File

@ -80,6 +80,10 @@ ws.send('listening.generate', {}, null, (p) => {
- **No client-side timeout.** If the backend never emits `<event>: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).

View File

@ -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 |

4
package-lock.json generated
View File

@ -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",

View File

@ -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",

View File

@ -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.

148
src/ws.ts
View File

@ -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<WsSocket | void> | null = null
// A connect() that landed on the reconnect throttle is retried when the
// throttle window opens (instead of being dropped).
private _deferredConnectTimer: ReturnType<typeof setTimeout> | null = null
// Auto-reconnect scheduled by a close; destroy() cancels it.
private _reconnectTimer: ReturnType<typeof setTimeout> | null = null
// rejectOnDisconnect: sends written to a socket, keyed by that socket.
private _inflight = new Set<InflightSend>()
private _payloadSend = new WeakMap<object, InflightSend>()
constructor(config: WsClientConfig, createRef: CreateRefFn) {
super()
this._config = {
@ -266,12 +283,13 @@ class WsClientImpl extends EventTarget implements WsClient {
ensureConnected(): Promise<void> {
if (this.socket?.socket_id && this.is_opened.value) return Promise.resolve()
if (!this._connectedPromise) {
this._connectedPromise = new Promise<void>((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<void>((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<T>
}
/** 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()

214
test/ws.test.mjs Normal file
View File

@ -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()')
})