1+import fs from "node:fs/promises";
2+import { type AddressInfo, Socket } from "node:net";
3+import { tmpdir } from "node:os";
4+import path from "node:path";
5+import { afterEach, describe, expect, it } from "vitest";
6+import { type RawData, WebSocketServer } from "ws";
7+import { probeGateway } from "./probe.js";
8+import { PROTOCOL_VERSION } from "./protocol/index.js";
9+10+const tempRoots: string[] = [];
11+12+function rawWsDataToString(data: RawData): string {
13+if (typeof data === "string") {
14+return data;
15+}
16+if (Buffer.isBuffer(data)) {
17+return data.toString("utf8");
18+}
19+if (Array.isArray(data)) {
20+return Buffer.concat(data).toString("utf8");
21+}
22+return Buffer.from(data).toString("utf8");
23+}
24+25+function activeClientSocketsToPort(port: number): Socket[] {
26+// Node has no public active-handle API; this regression must prove the probe
27+// promise does not resolve while the client-side socket handle is still live.
28+// oxlint-disable no-underscore-dangle
29+const handles =
30+(process as typeof process & { _getActiveHandles?: () => unknown[] })._getActiveHandles?.() ??
31+[];
32+// oxlint-enable no-underscore-dangle
33+return handles.filter(
34+(handle): handle is Socket => handle instanceof Socket && handle.remotePort === port,
35+);
36+}
37+38+async function createTempStateDir(): Promise<string> {
39+const tempRoot = await fs.mkdtemp(path.join(tmpdir(), "openclaw-probe-close-drain-"));
40+tempRoots.push(tempRoot);
41+return tempRoot;
42+}
43+44+describe("probeGateway close drain", () => {
45+afterEach(async () => {
46+await Promise.all(
47+tempRoots.splice(0).map((root) => fs.rm(root, { recursive: true, force: true })),
48+);
49+});
50+51+it("waits for the real WebSocket client socket to close before resolving", async () => {
52+const wss = new WebSocketServer({ host: "127.0.0.1", port: 0 });
53+let sawConnect = false;
54+55+wss.on("connection", (ws) => {
56+ws.send(
57+JSON.stringify({
58+type: "event",
59+event: "connect.challenge",
60+payload: { nonce: "test-nonce" },
61+}),
62+);
63+ws.on("message", (raw) => {
64+const frame = JSON.parse(rawWsDataToString(raw)) as {
65+id?: string;
66+method?: string;
67+type?: string;
68+};
69+if (frame.type !== "req" || frame.method !== "connect" || !frame.id) {
70+return;
71+}
72+sawConnect = true;
73+ws.send(
74+JSON.stringify({
75+type: "res",
76+id: frame.id,
77+ok: true,
78+payload: {
79+type: "hello-ok",
80+protocol: PROTOCOL_VERSION,
81+server: { version: "test", connId: "conn-test" },
82+features: { methods: [], events: ["connect.challenge"] },
83+snapshot: {},
84+auth: { role: "operator", scopes: ["operator.read"] },
85+policy: {
86+maxPayload: 1_000_000,
87+maxBufferedBytes: 1_000_000,
88+tickIntervalMs: 60_000,
89+},
90+},
91+}),
92+);
93+});
94+});
95+96+try {
97+await new Promise<void>((resolve) => wss.once("listening", resolve));
98+const port = (wss.address() as AddressInfo).port;
99+100+const result = await probeGateway({
101+url: `ws://127.0.0.1:${port}`,
102+auth: { token: "secret" },
103+timeoutMs: 1_000,
104+includeDetails: false,
105+env: { ...process.env, OPENCLAW_STATE_DIR: await createTempStateDir() },
106+});
107+108+expect(result.ok).toBe(true);
109+expect(sawConnect).toBe(true);
110+expect(activeClientSocketsToPort(port)).toHaveLength(0);
111+} finally {
112+for (const client of wss.clients) {
113+client.terminate();
114+}
115+await new Promise<void>((resolve, reject) => {
116+wss.close((error) => (error ? reject(error) : resolve()));
117+});
118+}
119+});
120+});
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。