Skip to content

Commit 5aa7c62

Browse files
zhangguiping-xydtzhangguiping-xydt
authored
fix(irc): monitor stays disconnected after IRC socket closes (#100799)
1 parent 6418e19 commit 5aa7c62

4 files changed

Lines changed: 397 additions & 78 deletions

File tree

‎extensions/irc/src/client.test.ts‎

Lines changed: 98 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,14 @@ type LoopbackIrcServer = {
99
close(): Promise<void>;
1010
};
1111

12+
type HangingIrcServer = {
13+
port: number;
14+
acceptedCount: number;
15+
closedCount: number;
16+
openSocketCount(): number;
17+
close(): Promise<void>;
18+
};
19+
1220
async function startLoopbackIrcServer(): Promise<LoopbackIrcServer> {
1321
const lines: string[] = [];
1422
const sockets = new Set<net.Socket>();
@@ -60,6 +68,69 @@ async function startLoopbackIrcServer(): Promise<LoopbackIrcServer> {
6068
};
6169
}
6270

71+
async function waitForIrcCondition(
72+
predicate: () => boolean,
73+
message: string,
74+
timeoutMs = 1000,
75+
): Promise<void> {
76+
const deadline = Date.now() + timeoutMs;
77+
while (!predicate()) {
78+
if (Date.now() >= deadline) {
79+
throw new Error(message);
80+
}
81+
await new Promise((resolve) => {
82+
setTimeout(resolve, 10);
83+
});
84+
}
85+
}
86+
87+
async function startHangingIrcServer(): Promise<HangingIrcServer> {
88+
const sockets = new Set<net.Socket>();
89+
let acceptedCount = 0;
90+
let closedCount = 0;
91+
const server = net.createServer((socket) => {
92+
acceptedCount += 1;
93+
sockets.add(socket);
94+
socket.setEncoding("utf8");
95+
socket.on("data", () => {});
96+
socket.on("close", () => {
97+
sockets.delete(socket);
98+
closedCount += 1;
99+
});
100+
});
101+
await new Promise<void>((resolve) => {
102+
server.listen(0, "127.0.0.1", resolve);
103+
});
104+
const address = server.address();
105+
if (!address || typeof address === "string") {
106+
throw new Error("expected loopback IRC server to bind a TCP port");
107+
}
108+
return {
109+
port: address.port,
110+
get acceptedCount() {
111+
return acceptedCount;
112+
},
113+
get closedCount() {
114+
return closedCount;
115+
},
116+
openSocketCount: () => sockets.size,
117+
close: async () => {
118+
for (const socket of sockets) {
119+
socket.destroy();
120+
}
121+
await new Promise<void>((resolve, reject) => {
122+
server.close((error) => {
123+
if (error) {
124+
reject(error);
125+
return;
126+
}
127+
resolve();
128+
});
129+
});
130+
},
131+
};
132+
}
133+
63134
describe("irc client nickserv", () => {
64135
it("builds IDENTIFY command when password is set", () => {
65136
expect(
@@ -101,6 +172,33 @@ describe("irc client nickserv", () => {
101172
});
102173
});
103174

175+
describe("irc client readiness timeout", () => {
176+
it("closes the socket when registration never becomes ready", async () => {
177+
const server = await startHangingIrcServer();
178+
try {
179+
await expect(
180+
connectIrcClient({
181+
host: "127.0.0.1",
182+
port: server.port,
183+
tls: false,
184+
nick: "bot",
185+
username: "bot",
186+
realname: "OpenClaw Bot",
187+
connectTimeoutMs: 50,
188+
}),
189+
).rejects.toThrow(/IRC connect/);
190+
191+
expect(server.acceptedCount).toBeGreaterThanOrEqual(1);
192+
await waitForIrcCondition(
193+
() => server.closedCount >= 1 && server.openSocketCount() === 0,
194+
`expected timed-out IRC connect socket to close; accepted=${server.acceptedCount} closed=${server.closedCount} open=${server.openSocketCount()}`,
195+
);
196+
} finally {
197+
await server.close();
198+
}
199+
});
200+
});
201+
104202
describe("irc client fallback nick", () => {
105203
it("produces unique fallback nicks across sequential calls", () => {
106204
const first = buildFallbackNick("bot");

‎extensions/irc/src/client.ts‎

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -69,6 +69,7 @@ export type IrcClientOptions = {
6969
onPrivmsg?: (event: IrcPrivmsgEvent) => void | Promise<void>;
7070
onNotice?: (text: string, target?: string) => void;
7171
onError?: (error: Error) => void;
72+
onDisconnect?: () => void;
7273
onLine?: (line: string) => void;
7374
};
7475

@@ -430,8 +431,12 @@ export async function connectIrcClient(options: IrcClientOptions): Promise<IrcCl
430431
socket.once("close", () => {
431432
if (!closed) {
432433
closed = true;
434+
removeAbortListener?.();
435+
removeAbortListener = null;
433436
if (!ready) {
434437
fail(new Error("IRC connection closed before ready"));
438+
} else {
439+
options.onDisconnect?.();
435440
}
436441
}
437442
});
@@ -452,7 +457,12 @@ export async function connectIrcClient(options: IrcClientOptions): Promise<IrcCl
452457
}
453458
}
454459

455-
await withTimeout(readyPromise, timeoutMs, "IRC connect");
460+
try {
461+
await withTimeout(readyPromise, timeoutMs, "IRC connect");
462+
} catch (error) {
463+
close();
464+
throw error;
465+
}
456466

457467
return {
458468
get nick() {

‎extensions/irc/src/monitor.test.ts‎

Lines changed: 149 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,153 @@
11
// Irc tests cover monitor plugin behavior.
2-
import { describe, expect, it } from "vitest";
3-
import { resolveIrcInboundTarget } from "./monitor.js";
2+
import net from "node:net";
3+
import { afterEach, describe, expect, it, vi } from "vitest";
4+
import { monitorIrcProvider, resolveIrcInboundTarget } from "./monitor.js";
5+
import { clearIrcRuntime, setIrcRuntime } from "./runtime.js";
6+
import type { CoreConfig } from "./types.js";
7+
8+
type DisconnectingIrcServer = {
9+
port: number;
10+
lines: string[];
11+
connectionCount: number;
12+
close(): Promise<void>;
13+
};
14+
15+
async function waitForIrcCondition(
16+
predicate: () => boolean,
17+
message: string,
18+
timeoutMs = 3000,
19+
): Promise<void> {
20+
const deadline = Date.now() + timeoutMs;
21+
while (!predicate()) {
22+
if (Date.now() >= deadline) {
23+
throw new Error(message);
24+
}
25+
await new Promise((resolve) => {
26+
setTimeout(resolve, 10);
27+
});
28+
}
29+
}
30+
31+
async function startDisconnectingIrcServer(): Promise<DisconnectingIrcServer> {
32+
const lines: string[] = [];
33+
const sockets = new Set<net.Socket>();
34+
let connectionCount = 0;
35+
36+
const server = net.createServer((socket) => {
37+
const connectionNumber = ++connectionCount;
38+
sockets.add(socket);
39+
socket.setEncoding("utf8");
40+
let buffer = "";
41+
socket.on("data", (chunk: string) => {
42+
buffer += chunk;
43+
let idx = buffer.indexOf("\n");
44+
while (idx !== -1) {
45+
const line = buffer.slice(0, idx).replace(/\r$/, "");
46+
buffer = buffer.slice(idx + 1);
47+
idx = buffer.indexOf("\n");
48+
lines.push(line);
49+
if (line.startsWith("USER ")) {
50+
socket.write(":server 001 bot :welcome\r\n");
51+
if (connectionNumber === 1) {
52+
setTimeout(() => socket.destroy(), 10);
53+
}
54+
}
55+
}
56+
});
57+
socket.on("close", () => {
58+
sockets.delete(socket);
59+
});
60+
});
61+
62+
await new Promise<void>((resolve) => {
63+
server.listen(0, "127.0.0.1", resolve);
64+
});
65+
const address = server.address();
66+
if (!address || typeof address === "string") {
67+
throw new Error("expected loopback IRC server to bind a TCP port");
68+
}
69+
70+
return {
71+
port: address.port,
72+
lines,
73+
get connectionCount() {
74+
return connectionCount;
75+
},
76+
close: async () => {
77+
for (const socket of sockets) {
78+
socket.destroy();
79+
}
80+
await new Promise<void>((resolve, reject) => {
81+
server.close((error) => {
82+
if (error) {
83+
reject(error);
84+
return;
85+
}
86+
resolve();
87+
});
88+
});
89+
},
90+
};
91+
}
92+
93+
function installMonitorRuntime() {
94+
setIrcRuntime({
95+
logging: {
96+
shouldLogVerbose: vi.fn(() => false),
97+
getChildLogger: vi.fn(() => ({
98+
debug: vi.fn(),
99+
info: vi.fn(),
100+
warn: vi.fn(),
101+
error: vi.fn(),
102+
})),
103+
},
104+
channel: {
105+
activity: {
106+
record: vi.fn(),
107+
},
108+
},
109+
} as never);
110+
}
111+
112+
afterEach(() => {
113+
clearIrcRuntime();
114+
});
115+
116+
describe("irc monitor reconnect", () => {
117+
it("reconnects when an established IRC socket closes", async () => {
118+
installMonitorRuntime();
119+
const server = await startDisconnectingIrcServer();
120+
const config = {
121+
channels: {
122+
irc: {
123+
host: "127.0.0.1",
124+
port: server.port,
125+
tls: false,
126+
nick: "bot",
127+
username: "bot",
128+
realname: "OpenClaw",
129+
channels: ["#openclaw"],
130+
},
131+
},
132+
} as CoreConfig;
133+
let monitor: { stop: () => void } | undefined;
134+
135+
try {
136+
monitor = await monitorIrcProvider({ config });
137+
await waitForIrcCondition(
138+
() =>
139+
server.connectionCount >= 2 &&
140+
server.lines.filter((line) => line === "USER bot 0 * :OpenClaw").length >= 2,
141+
"expected IRC monitor to reconnect after the first socket closed",
142+
);
143+
144+
expect(server.connectionCount).toBeGreaterThanOrEqual(2);
145+
} finally {
146+
monitor?.stop();
147+
await server.close();
148+
}
149+
});
150+
});
4151

5152
describe("irc monitor inbound target", () => {
6153
it("keeps channel target for group messages", () => {

0 commit comments

Comments
 (0)