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}`);