From 87f7bd8c5456664151144d9d26ea801f428f94bb Mon Sep 17 00:00:00 2001 From: DrSmoothl <1787882683@qq.com> Date: Sat, 7 Nov 2026 12:57:25 +0800 Subject: [PATCH] fix(websocket): settle failed sends and enforce SDK request deadlines Settle queued sends once when a connection fails or closes before opening, and reject further sends while preserving the original failure. Include connection waits in SDK request deadlines without closing shared connections or sending expired actions after a late connection. Refs #665 Refs #666 --- packages/sdk/CHANGELOG.md | 2 + packages/sdk/README.md | 3 + packages/sdk/src/client/websocket-client.ts | 155 +++++++++-------- packages/sdk/src/types/json.ts | 4 +- packages/sdk/tests/websocket-client.test.ts | 183 ++++++++++++++++++++ packages/websocket/src/client.ts | 92 +++++++--- packages/websocket/tests/client.test.ts | 106 ++++++++++++ 7 files changed, 442 insertions(+), 103 deletions(-) diff --git a/packages/sdk/CHANGELOG.md b/packages/sdk/CHANGELOG.md index 2854cf39..8a22ee94 100644 --- a/packages/sdk/CHANGELOG.md +++ b/packages/sdk/CHANGELOG.md @@ -2,6 +2,8 @@ ## Unreleased +- Fix ordinary WebSocket request deadlines to include connection waits, prevent expired requests from being sent after a late open, and preserve shared connections when an individual request times out or is cancelled. Refs #666. + - Add typed public contact profiles and opt-in friend-list enrichment via `getFriendList({ withProfile: true })` or `with_profile` on either list action; existing request options remain compatible. See [contact profiles](../onebot/CONTACT_PROFILES.md). - Add a Node reverse WebSocket server with authenticated per-account sessions, Universal and paired API/Event channels, existing action/event/stream helpers, bounded buffers and explicit disconnect handling; see [REVERSE_WEBSOCKET.md](REVERSE_WEBSOCKET.md). diff --git a/packages/sdk/README.md b/packages/sdk/README.md index c3752329..2558bd90 100644 --- a/packages/sdk/README.md +++ b/packages/sdk/README.md @@ -375,6 +375,9 @@ await status; ``` `timeoutMs: 0` 表示关闭该请求的超时控制。 +WebSocket 普通请求的超时从调用时开始，包含等待连接建立和等待响应的总时间， +连接成功后不会重新计时。等待连接时已超时或取消的请求不会在连接晚到后补发； +取消单个请求不会关闭其他请求正在使用或等待的共享连接。 ## 错误处理 diff --git a/packages/sdk/src/client/websocket-client.ts b/packages/sdk/src/client/websocket-client.ts index 7dcbf179..08bab75d 100644 --- a/packages/sdk/src/client/websocket-client.ts +++ b/packages/sdk/src/client/websocket-client.ts @@ -512,91 +512,94 @@ export class SnowLumaWebSocketClient extends SnowLumaApiClient { params?: ActionParams, options?: RequestOptions, ): Promise>> { - throwIfAborted(options?.signal); - await raceWithAbort( - this.connect(), - options?.signal, - 'SnowLuma WebSocket request aborted before connection opened', - ); - throwIfAborted(options?.signal); - - const socket = this.connection?.socket; - - if (!socket || socket.readyState !== WEBSOCKET_OPEN) { - throw new SnowLumaConnectionError('SnowLuma WebSocket is not open'); - } - - const echo = options?.echo ?? createEcho(); - const key = echoKey(echo); const timeoutMs = options?.timeoutMs ?? this.requestTimeoutMs; + const abort = createRequestAbortState(timeoutMs, options?.signal); - if (this.pending.has(key)) { - throw new SnowLumaConnectionError( - 'A WebSocket request with this echo is already pending', + try { + throwIfAborted(abort.signal); + await raceWithAbort( + this.connect(), + abort.signal, + 'SnowLuma WebSocket request aborted before connection opened', ); - } + throwIfAborted(abort.signal); - return new Promise>>((resolve, reject) => { - const cleanup = (): void => { - this.pending.delete(key); + const socket = this.connection?.socket; - if (timer) { - clearTimeout(timer); - } + if (!socket || socket.readyState !== WEBSOCKET_OPEN) { + throw new SnowLumaConnectionError('SnowLuma WebSocket is not open'); + } - options?.signal?.removeEventListener('abort', abortListener); - }; + const echo = options?.echo ?? createEcho(); + const key = echoKey(echo); - const pending: PendingRequest = { - resolve: (response) => { - cleanup(); - resolve(response as ApiResponse>); - }, - reject: (error) => { - cleanup(); - reject(error); - }, - }; - - const timer = - timeoutMs > 0 - ? setTimeout(() => { - pending.reject( - new SnowLumaTimeoutError( - `SnowLuma WebSocket request timed out after ${timeoutMs}ms`, - timeoutMs, - ), - ); - }, timeoutMs) - : null; - - const abortListener = () => { - pending.reject( - new SnowLumaAbortError('SnowLuma WebSocket request aborted', { - cause: options?.signal?.reason, - }), - ); - }; - - this.pending.set(key, pending); - options?.signal?.addEventListener('abort', abortListener, { once: true }); - - try { - socket.send( - JSON.stringify({ - action, - params: toJsonObject(params as JsonObject | undefined), - echo, - }), - ); - } catch (error) { - pending.reject( - new SnowLumaConnectionError('Failed to send SnowLuma WebSocket request', { - cause: error, - }), + if (this.pending.has(key)) { + throw new SnowLumaConnectionError( + 'A WebSocket request with this echo is already pending', ); } - }); + + return await new Promise>>((resolve, reject) => { + const cleanup = (): void => { + if (this.pending.get(key) === pending) { + this.pending.delete(key); + } + + abort.signal.removeEventListener('abort', abortListener); + abort.cleanup(); + }; + + const pending: PendingRequest = { + resolve: (response) => { + cleanup(); + resolve(response as ApiResponse>); + }, + reject: (error) => { + cleanup(); + reject(error); + }, + }; + + const abortListener = (): void => { + pending.reject( + new SnowLumaAbortError('SnowLuma WebSocket request aborted', { + cause: abort.signal.reason, + }), + ); + }; + + this.pending.set(key, pending); + abort.signal.addEventListener('abort', abortListener, { once: true }); + + try { + socket.send( + JSON.stringify({ + action, + params: toJsonObject(params as JsonObject | undefined), + echo, + }), + ); + } catch (error) { + pending.reject( + new SnowLumaConnectionError('Failed to send SnowLuma WebSocket request', { + cause: error, + }), + ); + } + }); + } catch (error) { + if (abort.isTimedOut()) { + throw new SnowLumaTimeoutError( + `SnowLuma WebSocket request timed out after ${timeoutMs}ms`, + timeoutMs, + { cause: error }, + ); + } + + throw error; + } finally { + abort.cleanup(); + } } override async *requestStream( diff --git a/packages/sdk/src/types/json.ts b/packages/sdk/src/types/json.ts index 378410e2..1b0589e4 100644 --- a/packages/sdk/src/types/json.ts +++ b/packages/sdk/src/types/json.ts @@ -28,7 +28,7 @@ export type MessageFormat = 'array' | 'string'; export interface RequestOptions { /** Echo value used to correlate OneBot request and response packets. */ echo?: JsonValue; - /** Override the client's default timeout for this request. Set to 0 to disable timeout. */ + /** Total request timeout, including connection and response waits. Set to 0 to disable. */ timeoutMs?: number; /** Cancels the request before a SnowLuma response arrives. */ signal?: AbortSignal; @@ -38,6 +38,6 @@ export interface RequestOptions { export interface SnowLumaClientOptions { /** Access token configured in SnowLuma's OneBot settings. */ accessToken?: string; - /** Default timeout for API calls in milliseconds. Defaults to 30000. */ + /** Default total timeout for API calls in milliseconds, including connection waits. Defaults to 30000. */ requestTimeoutMs?: number; } diff --git a/packages/sdk/tests/websocket-client.test.ts b/packages/sdk/tests/websocket-client.test.ts index 2c3dcdeb..a62a0042 100644 --- a/packages/sdk/tests/websocket-client.test.ts +++ b/packages/sdk/tests/websocket-client.test.ts @@ -302,6 +302,189 @@ describe('SnowLumaWebSocketClient', () => { vi.useRealTimers(); }); + it('times out while opening and never sends the expired action after a late open', async () => { + vi.useFakeTimers(); + const client = new SnowLumaWebSocketClient({ webSocket: MockWebSocket }); + const settled = vi.fn(); + const request = client.getStatus({ timeoutMs: 20 }).then(settled, settled); + const socket = MockWebSocket.instances.at(-1)!; + + try { + await vi.advanceTimersByTimeAsync(20); + expect(settled).toHaveBeenCalledExactlyOnceWith(expect.any(SnowLumaTimeoutError)); + expect(settled.mock.calls[0]![0]).toHaveProperty('timeoutMs', 20); + socket.open(); + await vi.advanceTimersByTimeAsync(0); + expect(socket.sent).toHaveLength(0); + expect(client.isConnected).toBe(true); + expect(vi.getTimerCount()).toBe(0); + } finally { + client.close(); + await request; + } + }); + + it('uses the client default as one deadline across connection and response waits', async () => { + vi.useFakeTimers(); + const client = new SnowLumaWebSocketClient({ + webSocket: MockWebSocket, + requestTimeoutMs: 30, + }); + const settled = vi.fn(); + const request = client.getStatus({ echo: 'deadline' }).then(settled, settled); + const socket = MockWebSocket.instances.at(-1)!; + + try { + await vi.advanceTimersByTimeAsync(20); + socket.open(); + await vi.advanceTimersByTimeAsync(0); + expect(socket.sent).toHaveLength(1); + await vi.advanceTimersByTimeAsync(9); + expect(settled).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(1); + expect(settled).toHaveBeenCalledExactlyOnceWith(expect.any(SnowLumaTimeoutError)); + expect(settled.mock.calls[0]![0]).toHaveProperty('timeoutMs', 30); + expect(vi.getTimerCount()).toBe(0); + } finally { + client.close(); + await request; + } + }); + + it.each(['timeout', 'abort'] as const)( + 'keeps a shared opening connection usable after one request ends by %s', + async (outcome) => { + vi.useFakeTimers(); + const client = new SnowLumaWebSocketClient({ webSocket: MockWebSocket }); + const controller = new AbortController(); + const remove = vi.spyOn(controller.signal, 'removeEventListener'); + const firstSettled = vi.fn(); + const first = client + .getStatus({ echo: 'first', timeoutMs: 20, signal: controller.signal }) + .then(firstSettled, firstSettled); + const socket = MockWebSocket.instances.at(-1)!; + const second = client.getStatus({ echo: 'second', timeoutMs: 100 }); + const secondResult = second.catch((error: unknown) => error); + + try { + if (outcome === 'abort') { + controller.abort('caller cancelled'); + } + + await vi.advanceTimersByTimeAsync(20); + expect(firstSettled).toHaveBeenCalledExactlyOnceWith( + expect.any(outcome === 'timeout' ? SnowLumaTimeoutError : SnowLumaAbortError), + ); + + if (outcome === 'abort') { + expect(firstSettled.mock.calls[0]![0]).toHaveProperty( + 'cause', + 'caller cancelled', + ); + } + + expect(remove).toHaveBeenCalledWith('abort', expect.any(Function)); + expect(socket.readyState).toBe(0); + socket.open(); + await vi.advanceTimersByTimeAsync(0); + expect(socket.sent.map((value) => JSON.parse(value).echo)).toEqual(['second']); + socket.message( + JSON.stringify({ + status: 'ok', + retcode: 0, + data: { online: true }, + echo: 'second', + }), + ); + await expect(second).resolves.toEqual({ online: true }); + expect(client.isConnected).toBe(true); + expect(vi.getTimerCount()).toBe(0); + } finally { + client.close(); + await Promise.all([first, secondResult]); + } + }, + ); + + it('allows zero timeout through connection and response waits', async () => { + vi.useFakeTimers(); + const client = new SnowLumaWebSocketClient({ + webSocket: MockWebSocket, + requestTimeoutMs: 10, + }); + const settled = vi.fn(); + const request = client + .getStatus({ timeoutMs: 0, echo: 'unlimited' }) + .then(settled, settled); + const socket = MockWebSocket.instances.at(-1)!; + + try { + await vi.advanceTimersByTimeAsync(100); + expect(settled).not.toHaveBeenCalled(); + socket.open(); + await vi.advanceTimersByTimeAsync(100); + expect(settled).not.toHaveBeenCalled(); + expect(socket.sent).toHaveLength(1); + socket.message( + JSON.stringify({ + status: 'ok', + retcode: 0, + data: { online: true }, + echo: 'unlimited', + }), + ); + await request; + expect(settled).toHaveBeenCalledExactlyOnceWith({ online: true }); + expect(vi.getTimerCount()).toBe(0); + } finally { + client.close(); + await request; + } + }); + + it.each(['close', 'constructor failure', 'pre-aborted'] as const)( + 'cleans request controls when opening ends with %s', + async (outcome) => { + vi.useFakeTimers(); + const failure = new Error('opening failed'); + class OpeningSocket extends MockWebSocket { + constructor(url: string) { + if (outcome === 'constructor failure') { + throw failure; + } + + super(url); + } + } + const client = new SnowLumaWebSocketClient({ webSocket: OpeningSocket }); + const controller = new AbortController(); + const remove = vi.spyOn(controller.signal, 'removeEventListener'); + + if (outcome === 'pre-aborted') { + controller.abort(failure); + } + + const before = MockWebSocket.instances.length; + const result = client + .getStatus({ timeoutMs: 20, signal: controller.signal }) + .catch((error: unknown) => error); + + if (outcome === 'close') { + MockWebSocket.instances.at(-1)!.close(); + expect(await result).toBeInstanceOf(SnowLumaConnectionError); + } else if (outcome === 'constructor failure') { + expect(await result).toBe(failure); + } else { + expect(await result).toBeInstanceOf(SnowLumaAbortError); + expect(MockWebSocket.instances).toHaveLength(before); + } + + expect(vi.getTimerCount()).toBe(0); + expect(remove).toHaveBeenCalledWith('abort', expect.any(Function)); + client.close(); + }, + ); + it('rejects duplicate in-flight echoes without replacing the original request', async () => { const client = new SnowLumaWebSocketClient({ webSocket: MockWebSocket }); const opening = client.connect(); diff --git a/packages/websocket/src/client.ts b/packages/websocket/src/client.ts index 75425663..6d23a32f 100644 --- a/packages/websocket/src/client.ts +++ b/packages/websocket/src/client.ts @@ -15,6 +15,7 @@ import { WebSocket as InternalWebSocket, CONNECTING, OPEN, + CLOSING, CLOSED, isValidCloseCode, type SendCallback, @@ -137,6 +138,10 @@ export class WebSocketClient extends EventEmitter { } as net.NetConnectOpts & tls.ConnectionOptions; const onConnect = () => { + if (this._readyState !== CONNECTING) { + return; + } + const pathAndQuery = (this._url.pathname || '/') + (this._url.search || ''); const hostHeader = this._url.port && Number(this._url.port) !== (this._secure ? 443 : 80) @@ -194,6 +199,10 @@ export class WebSocketClient extends EventEmitter { } private _onHandshakeData(chunk: Buffer): void { + if (this._readyState !== CONNECTING) { + return; + } + this._handshakeBuf = Buffer.concat([this._handshakeBuf, chunk]); const idx = this._handshakeBuf.indexOf('\r\n\r\n'); @@ -256,6 +265,7 @@ export class WebSocketClient extends EventEmitter { this._socket.removeAllListeners('error'); this._socket.removeAllListeners('close'); + this._handshakeBuf = Buffer.alloc(0); this._readyState = OPEN; this._ws = new InternalWebSocket(this._socket, { isServer: false, @@ -279,36 +289,65 @@ export class WebSocketClient extends EventEmitter { this._ws._onData(rest); } - for (const entry of this._pendingQueue) { - this._ws.send(entry.data, entry.options, entry.cb); + if (this._ws.readyState !== OPEN) { + this._failPendingSends(new Error('WebSocket is not open')); + return; } + const pending = this._pendingQueue; this._pendingQueue = []; + + for (const entry of pending) { + this._ws.send(entry.data, entry.options, entry.cb); + } } private _onEarlyError(err: Error): void { + if (this._ws || this._readyState === CLOSED) { + return; + } + + this._stopHandshake(err); this.emit('error', err); } private _onEarlyClose(): void { - if (!this._ws) { - this._readyState = CLOSED; - this.emit('close', 1006, ''); + if (this._ws || this._readyState === CLOSED) { + return; + } + + this._stopHandshake(new Error('WebSocket closed before handshake completed')); + this._readyState = CLOSED; + this.emit('close', 1006, ''); + } + + private _failPendingSends(err: Error): void { + const pending = this._pendingQueue; + this._pendingQueue = []; + + // Release queued payloads before invoking user code. Each callback is + // scheduled independently so reentrant sends cannot rejoin this queue. + for (const entry of pending) { + if (entry.cb) { + process.nextTick(entry.cb, err); + } } } - private _abort(msg: string): void { - const err = new Error('WebSocket handshake failed: ' + msg); - - try { - this._socket.destroy(); - } catch { - /* noop */ + private _stopHandshake(err: Error): void { + if (this._readyState !== CONNECTING) { + return; } - this.emit('error', err); - this._readyState = CLOSED; - this.emit('close', 1006, ''); + this._readyState = CLOSING; + this._handshakeBuf = Buffer.alloc(0); + this._failPendingSends(err); + this._socket.destroy(); + } + + private _abort(msg: string): void { + this._onEarlyError(new Error('WebSocket handshake failed: ' + msg)); + this._onEarlyClose(); } send( @@ -325,6 +364,17 @@ export class WebSocketClient extends EventEmitter { options = undefined; } + if (this._readyState !== CONNECTING) { + const err = new Error('WebSocket is not open'); + + if (cb) { + cb(err); + return; + } + + throw err; + } + this._pendingQueue.push({ data, options, cb }); } @@ -344,11 +394,7 @@ export class WebSocketClient extends EventEmitter { return this._ws.close(code, reason); } - try { - this._socket.destroy(); - } catch { - /* noop */ - } + this._stopHandshake(new Error('WebSocket closed before handshake completed')); } terminate(): void { @@ -356,10 +402,6 @@ export class WebSocketClient extends EventEmitter { return this._ws.terminate(); } - try { - this._socket.destroy(); - } catch { - /* noop */ - } + this._stopHandshake(new Error('WebSocket terminated before handshake completed')); } } diff --git a/packages/websocket/tests/client.test.ts b/packages/websocket/tests/client.test.ts index c634c19c..37ff0fd8 100644 --- a/packages/websocket/tests/client.test.ts +++ b/packages/websocket/tests/client.test.ts @@ -875,6 +875,112 @@ describe('WebSocketClient', () => { expect(second.payload).toEqual(Buffer.from('bin')); }); + it('settles queued sends once on a rejected handshake and rejects further sends', async () => { + const { port, nextSocket } = await listenPlain(); + const client = createClient(`ws://127.0.0.1:${port}`); + const events = track(client); + const reentrant = vi.fn(); + const fromError = vi.fn(); + const first = vi.fn(() => client.send('from failed callback', reentrant)); + const second = vi.fn(); + client.on('error', () => client.send('from error listener', fromError)); + client.send('first', first); + client.send('second', { binary: true }, second); + client.send('without callback'); + const socket = await nextSocket(); + await readRequest(socket); + socket.end('HTTP/1.1 401 Unauthorized\r\nContent-Length: 0\r\n\r\n'); + await events.waitClose(); + await new Promise((resolve) => setImmediate(resolve)); + + expect(first).toHaveBeenCalledExactlyOnceWith(events.errors[0]); + expect(second).toHaveBeenCalledExactlyOnceWith(events.errors[0]); + expect(reentrant).toHaveBeenCalledExactlyOnceWith(expect.any(Error)); + expect(fromError).toHaveBeenCalledExactlyOnceWith(expect.any(Error)); + expect(events.errors[0]?.message).toContain('Unexpected response status 401'); + expect(events.closes).toEqual([{ code: 1006, reason: '' }]); + const afterClose = vi.fn(); + client.send('after close', afterClose); + expect(afterClose).toHaveBeenCalledExactlyOnceWith(expect.any(Error)); + expect(() => client.send('after close without callback')).toThrow('WebSocket is not open'); + + client.close(); + client.terminate(); + await new Promise((resolve) => setImmediate(resolve)); + expect(first).toHaveBeenCalledTimes(1); + expect(second).toHaveBeenCalledTimes(1); + expect(events.closes).toHaveLength(1); + }); + + it.each(['close', 'terminate', 'peer close'] as const)( + 'settles queued callbacks exactly once after %s before opening', + async (outcome) => { + const { port, nextSocket } = await listenPlain(); + const client = createClient(`ws://127.0.0.1:${port}`); + const events = track(client); + const first = vi.fn(); + const second = vi.fn(); + client.send('first', first); + client.send('second', { binary: true }, second); + client.send('without callback'); + const socket = await nextSocket(); + await readRequest(socket); + + if (outcome === 'peer close') { + socket.destroy(); + } else { + client[outcome](); + const whileClosing = vi.fn(); + client.send('while closing', whileClosing); + expect(whileClosing).toHaveBeenCalledExactlyOnceWith(expect.any(Error)); + } + + await events.waitClose(); + await new Promise((resolve) => setImmediate(resolve)); + expect(first).toHaveBeenCalledExactlyOnceWith(expect.any(Error)); + expect(second).toHaveBeenCalledExactlyOnceWith(expect.any(Error)); + expect(client.readyState).toBe(WebSocketClient.CLOSED); + expect(() => client.send('after close')).toThrow('WebSocket is not open'); + client.close(); + client.terminate(); + await new Promise((resolve) => setImmediate(resolve)); + expect(first).toHaveBeenCalledTimes(1); + expect(second).toHaveBeenCalledTimes(1); + expect(events.closes).toEqual([{ code: 1006, reason: '' }]); + }, + ); + + it('passes the original connection error to queued send callbacks', async () => { + const port = await unusedPort(); + const client = createClient(`ws://127.0.0.1:${port}`); + const events = track(client); + const callback = vi.fn(); + client.send('queued', callback); + await events.waitClose(); + expect(events.errors[0]).toHaveProperty('code', 'ECONNREFUSED'); + expect(callback).toHaveBeenCalledExactlyOnceWith(events.errors[0]); + expect(events.closes).toHaveLength(1); + }); + + it('settles the pre-open queue if an open listener immediately terminates the connection', async () => { + const { port, nextSocket } = await listenPlain(); + const client = createClient(`ws://127.0.0.1:${port}`); + const events = track(client); + const first = vi.fn(); + const second = vi.fn(); + client.send('without callback'); + client.send('first', first); + client.send('second', second); + client.once('open', () => client.terminate()); + const socket = await nextSocket(); + const req = await readRequest(socket); + socket.write(handshakeResponse(req.headers['sec-websocket-key']!)); + await events.waitClose(); + expect(first).toHaveBeenCalledExactlyOnceWith(expect.any(Error)); + expect(second).toHaveBeenCalledExactlyOnceWith(expect.any(Error)); + expect(events.closes).toHaveLength(1); + }); + it('delegates send/ping/pong/close after open and forwards control events', async () => { const { port, nextSocket } = await listenPlain(); const client = createClient(`ws://127.0.0.1:${port}`);