fix(test): harden gateway network ws timeout · openclaw/o...
vincentkoc
·
2026-05-27
·
via Recent Commits to openclaw:main
| Original file line number | Diff line number | Diff line change |
|---|
|
1 | 1 | import { WebSocket } from "ws"; |
2 | 2 | import { PROTOCOL_VERSION } from "../../../../dist/gateway/protocol/index.js"; |
| 3 | +import { waitForWebSocketOpen } from "./open-websocket.mjs"; |
3 | 4 | |
4 | 5 | const url = process.env.GW_URL; |
5 | 6 | const token = process.env.GW_TOKEN; |
@@ -24,20 +25,7 @@ function delay(ms) {
|
24 | 25 | |
25 | 26 | async function openSocket(timeoutMs = 10_000) { |
26 | 27 | const ws = new WebSocket(url); |
27 | | -await new Promise((resolve, reject) => { |
28 | | -const timer = setTimeout(() => { |
29 | | -ws.close(); |
30 | | -reject(new Error("ws open timeout")); |
31 | | -}, timeoutMs); |
32 | | -ws.once("open", () => { |
33 | | -clearTimeout(timer); |
34 | | -resolve(); |
35 | | -}); |
36 | | -ws.once("error", (error) => { |
37 | | -clearTimeout(timer); |
38 | | -reject(error instanceof Error ? error : new Error(String(error))); |
39 | | -}); |
40 | | -}); |
| 28 | +await waitForWebSocketOpen(ws, timeoutMs, "ws open timeout"); |
41 | 29 | return ws; |
42 | 30 | } |
43 | 31 | |
|
| Original file line number | Diff line number | Diff line change |
|---|
|
| 1 | +export function waitForWebSocketOpen(ws, timeoutMs, message = "ws open timeout") { |
| 2 | +return new Promise((resolve, reject) => { |
| 3 | +let settled = false; |
| 4 | + |
| 5 | +const settle = (fn, value) => { |
| 6 | +if (settled) { |
| 7 | +return; |
| 8 | +} |
| 9 | +settled = true; |
| 10 | +clearTimeout(timer); |
| 11 | +ws.off?.("open", onOpen); |
| 12 | +ws.off?.("error", onError); |
| 13 | +fn(value); |
| 14 | +}; |
| 15 | +const onOpen = () => settle(resolve); |
| 16 | +const onError = (error) => |
| 17 | +settle(reject, error instanceof Error ? error : new Error(String(error))); |
| 18 | +const timer = setTimeout(() => { |
| 19 | +const consumeAbortError = () => {}; |
| 20 | +const removeAbortErrorConsumer = () => { |
| 21 | +ws.off?.("error", consumeAbortError); |
| 22 | +ws.off?.("close", removeAbortErrorConsumer); |
| 23 | +}; |
| 24 | +try { |
| 25 | +ws.off?.("error", onError); |
| 26 | +ws.on?.("error", consumeAbortError); |
| 27 | +ws.once?.("close", removeAbortErrorConsumer); |
| 28 | +ws.terminate?.(); |
| 29 | +if (typeof ws.terminate !== "function") { |
| 30 | +ws.close?.(); |
| 31 | +} |
| 32 | +} finally { |
| 33 | +settle(reject, new Error(message)); |
| 34 | +} |
| 35 | +}, timeoutMs); |
| 36 | + |
| 37 | +timer.unref?.(); |
| 38 | +ws.once("open", onOpen); |
| 39 | +ws.once("error", onError); |
| 40 | +}); |
| 41 | +} |
| Original file line number | Diff line number | Diff line change |
|---|
|
| 1 | +import { EventEmitter } from "node:events"; |
| 2 | +import { describe, expect, it } from "vitest"; |
| 3 | +import { waitForWebSocketOpen } from "../../scripts/e2e/lib/gateway-network/open-websocket.mjs"; |
| 4 | + |
| 5 | +class FakeWebSocket extends EventEmitter { |
| 6 | +terminated = false; |
| 7 | +closed = false; |
| 8 | + |
| 9 | +terminate(): void { |
| 10 | +this.terminated = true; |
| 11 | +queueMicrotask(() => { |
| 12 | +this.emit("error", new Error("socket abort after terminate")); |
| 13 | +this.emit("close"); |
| 14 | +}); |
| 15 | +} |
| 16 | + |
| 17 | +close(): void { |
| 18 | +this.closed = true; |
| 19 | +} |
| 20 | +} |
| 21 | + |
| 22 | +describe("gateway network WebSocket open guard", () => { |
| 23 | +it("consumes abort errors after open timeouts", async () => { |
| 24 | +const ws = new FakeWebSocket(); |
| 25 | +const keepAlive = setTimeout(() => {}, 100); |
| 26 | + |
| 27 | +try { |
| 28 | +await expect(waitForWebSocketOpen(ws, 1)).rejects.toThrow("ws open timeout"); |
| 29 | +} finally { |
| 30 | +clearTimeout(keepAlive); |
| 31 | +} |
| 32 | +await new Promise((resolve) => setImmediate(resolve)); |
| 33 | + |
| 34 | +expect(ws.terminated).toBe(true); |
| 35 | +expect(ws.listenerCount("open")).toBe(0); |
| 36 | +expect(ws.listenerCount("error")).toBe(0); |
| 37 | +}); |
| 38 | + |
| 39 | +it("cleans listeners after successful opens", async () => { |
| 40 | +const ws = new FakeWebSocket(); |
| 41 | +const opened = waitForWebSocketOpen(ws, 100); |
| 42 | + |
| 43 | +ws.emit("open"); |
| 44 | + |
| 45 | +await expect(opened).resolves.toBeUndefined(); |
| 46 | +expect(ws.terminated).toBe(false); |
| 47 | +expect(ws.listenerCount("open")).toBe(0); |
| 48 | +expect(ws.listenerCount("error")).toBe(0); |
| 49 | +}); |
| 50 | +}); |
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。
