Skip to content

Commit 0c39f5a

Browse files
hobinjk-dtsclaude
andcommitted
Release every socket from the NB pressure registry when it closes for good
An NB socket leaves the process-wide pressure registry when its close surfaces, and the registry's sampler runs while it has any entry. Three ways of ending a socket for good never surfaced that close, so the entry, its socket and the sampler outlived it for the rest of the process: - ToolSocketServer.close() with a peer that lacks the session layer: the server told session peers __ts/bye but never closed a transport, and the WebSocket server's close() leaves its connections open, so a peer that is never told __ts/bye kept its connection. - ToolSocketServer.close() with a session held in grace: with no transport, nothing ever surfaced its close. - close() on a reconnecting NB socket: it kept its registry entry and gc timer, as if it were going to reconnect. ToolSocketServer.close() now closes each socket as its own close() does. That surfaces a grace session's close at once, running the application's close listeners inside the shutdown loop, so a listener that throws no longer stops the server closing the remaining sockets and itself: the loop finishes and rethrows the first error. _surfaceClose also runs the server's session bookkeeping in a finally, so a throwing listener cannot leave the server holding a closed socket. In a host that restarts its server in one process, e.g. a jest suite per server, the sampler kept writing to state on a global Symbol after jest had soft-deleted it, overflowing the stack in whichever suite ran next. The new tests drive NB directly, so the NB conformance config, which reruns the original suites against NB through index.nbshim.js (where NB.connect cannot load), leaves them out. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1 parent 0cb2887 commit 0c39f5a

5 files changed

Lines changed: 213 additions & 22 deletions

File tree

‎jest.nb.config.json‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,5 +4,9 @@
44
},
55
"testMatch": [
66
"**/src/**/*.test.js"
7+
],
8+
"testPathIgnorePatterns": [
9+
"/node_modules/",
10+
"/src/pressureRegistry\\.test\\.js$"
711
]
812
}

‎src/ToolSocket.js‎

Lines changed: 15 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -559,17 +559,21 @@ class ToolSocket {
559559
this._stopWatchdog();
560560
this.__ssn.releaseRetained();
561561
this.__ssn.clearGrace();
562-
this.triggerEvent('close', event);
563-
this.triggerEvent('disconnect', event);
564-
this.triggerEvent('status', this.readyState);
565-
// internal, after the application's listeners: server-side bookkeeping. A direct
566-
// hook rather than an event: an application may wipe a socket's listeners
567-
// (removeAllListeners — the cloud proxy does it to a superseded edge socket) and the
568-
// server must still forget the session when its grace runs out.
569-
if (this.server && this.server._onSessionClosed) {
570-
this.server._onSessionClosed(this);
571-
}
572-
this.triggerEvent('__ts:closed', event);
562+
try {
563+
this.triggerEvent('close', event);
564+
this.triggerEvent('disconnect', event);
565+
this.triggerEvent('status', this.readyState);
566+
} finally {
567+
// internal, after the application's listeners: server-side bookkeeping. A direct
568+
// hook rather than an event: an application may wipe a socket's listeners
569+
// (removeAllListeners — the cloud proxy does it to a superseded edge socket) and the
570+
// server must still forget the session when its grace runs out. In a finally for
571+
// the same reason: a listener that throws must not leave the server holding it.
572+
if (this.server && this.server._onSessionClosed) {
573+
this.server._onSessionClosed(this);
574+
}
575+
this.triggerEvent('__ts:closed', event);
576+
}
573577
}
574578

575579
/**

‎src/ToolSocketNB.js‎

Lines changed: 8 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -797,6 +797,11 @@ function enhance(ts, userOpts = {}) {
797797
// peer, same capability, same transfers — just tune the new raw socket and keep pumping
798798
ts.addEventListener('__ts:migrated', () => { tuneSocket(); sched.pump(); });
799799

800+
// Whether a close is the socket's last: an auto-reconnecting outgoing socket comes back
801+
// after a close, unless the application closed it (nb.userClosed, see below), while an
802+
// incoming socket (url === null) never does
803+
const closedForGood = () => !(opts.reconnect && ts.url) || nb.userClosed;
804+
800805
// enroll in the process-wide pressure registry (Node only)
801806
{
802807
const Pn = getPressure();
@@ -807,7 +812,7 @@ function enhance(ts, userOpts = {}) {
807812
sendHold: (h, sq) => sendInternal('__tsnb/hold', { h: h ? 1 : 0, s: sq, c: Pn.cfg.holdClass }, null, 0)
808813
};
809814
Pn._register(entry);
810-
ts.addEventListener('close', () => { if (!(opts.reconnect && ts.url)) Pn._unregister(entry); });
815+
ts.addEventListener('close', () => { if (closedForGood()) Pn._unregister(entry); });
811816
}
812817
}
813818

@@ -852,10 +857,9 @@ function enhance(ts, userOpts = {}) {
852857
// the timer list - a GC root - so without this the interval's closure
853858
// pins nb/sched/opts and the whole socket forever. Guarded exactly like
854859
// the pressure entry above: an auto-reconnecting outgoing socket keeps
855-
// its gc timer across reconnects, while an incoming socket (url === null)
856-
// never comes back and must release it.
860+
// its gc timer across reconnects, until it is closed for good.
857861
if (gc.unref) gc.unref();
858-
ts.addEventListener('close', () => { if (!(opts.reconnect && ts.url)) clearInterval(gc); });
862+
ts.addEventListener('close', () => { if (closedForGood()) clearInterval(gc); });
859863

860864
// ---------- public backpressure / flow-control API ----------
861865
ts.getBackpressure = () => sched.stats();

‎src/ToolSocketServer.js‎

Lines changed: 21 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -202,17 +202,31 @@ class ToolSocketServer {
202202
}
203203
this.infoSubscribers.clear();
204204
this.infoAutoEnabled.clear();
205-
// shutting down ends every session deliberately: tell each peer in-band so it does
206-
// not try to migrate, and surface every close at once (no grace)
207-
for (const toolSocket of this.sockets) {
208-
toolSocket.__ssn.userClosed = true;
209-
toolSocket.__ssn.clearGrace();
210-
if (toolSocket.__ssn.peer === true) {
211-
toolSocket._sendControl('__ts/bye', {});
205+
// shutting down ends every session deliberately, as close() on each socket does: a
206+
// session peer is told in-band so it does not try to migrate, the transport is closed
207+
// (the WebSocket server's close() leaves its connections open, so a peer that does not
208+
// answer __ts/bye kept its socket), and a session held in grace, with no transport,
209+
// surfaces its close at once. A close that never surfaces keeps the socket and what it
210+
// holds (e.g. NB's pressure registry entry) for good. A copy, since a surfacing close
211+
// takes its socket out of this.sockets.
212+
// A close surfaced at once runs the application's close listeners right here: one that
213+
// throws must not leave the rest of the sockets, and the server, open. Shutdown
214+
// finishes, then the first such error is rethrown to the caller
215+
let listenerError = null;
216+
for (const toolSocket of this.sockets.slice()) {
217+
try {
218+
toolSocket.close();
219+
} catch (error) {
220+
if (listenerError === null) {
221+
listenerError = error;
222+
}
212223
}
213224
}
214225
this.sessions.clear();
215226
this.server.close();
227+
if (listenerError !== null) {
228+
throw listenerError;
229+
}
216230
}
217231

218232
/**

‎src/pressureRegistry.test.js‎

Lines changed: 165 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,165 @@
1+
/* global test, expect, afterEach, jest */
2+
// NB's process-wide pressure registry (see ToolSocketNB.js) holds an entry per enhanced
3+
// socket and samples them on an interval while it has any. A socket leaves the registry when
4+
// its close surfaces, so every way an application ends a socket for good must surface one:
5+
// otherwise the entry, its socket, and the sampler outlive it for the rest of the process.
6+
const ToolSocket = require('./index.js');
7+
const NB = require('./ToolSocketNB.js');
8+
9+
jest.setTimeout(20000);
10+
11+
const sleep = ms => new Promise(resolve => setTimeout(resolve, ms));
12+
const waitFor = async (condition, timeoutMs = 5000, stepMs = 10) => {
13+
const start = Date.now();
14+
while (Date.now() - start < timeoutMs) {
15+
if (condition()) {
16+
return true;
17+
}
18+
await sleep(stepMs);
19+
}
20+
return condition();
21+
};
22+
23+
const registry = () => globalThis[Symbol.for('toolsocketnb.pressure.v1')];
24+
const registered = () => NB.pressure.state().sockets;
25+
26+
const running = [];
27+
afterEach(async () => {
28+
while (running.length) {
29+
try {
30+
running.pop().close();
31+
} catch (_e) { /* already closed */ }
32+
}
33+
await waitFor(() => registered() === 0);
34+
});
35+
36+
function startServer() {
37+
const server = new ToolSocket.Server({ port: 0 }, 'server');
38+
server.addEventListener('connection', socket => NB.enhance(socket, { reconnect: false }));
39+
running.push(server);
40+
return new Promise(resolve => server.server.on('listening', () => {
41+
resolve({ server, url: `ws://127.0.0.1:${server.server.address().port}` });
42+
}));
43+
}
44+
45+
function opened(socket) {
46+
return new Promise(resolve => socket.addEventListener('open', resolve));
47+
}
48+
49+
test('closing a server releases its connections from the registry', async () => {
50+
const { server, url } = await startServer();
51+
// a legacy client, as most of the edge server's own are, and an NB one
52+
const legacy = new ToolSocket(new URL(url), 'testnet', 'web');
53+
const nb = NB.connect(url, 'testnet', 'web', { reconnect: false });
54+
await Promise.all([opened(legacy), opened(nb)]);
55+
// the server's two connections and the NB client
56+
expect(await waitFor(() => registered() === 3)).toBe(true);
57+
expect(registry().timer).not.toBe(null);
58+
59+
const closed = [legacy, nb].map(socket => new Promise(resolve => socket.addEventListener('close', resolve)));
60+
server.close();
61+
await Promise.all(closed);
62+
expect(await waitFor(() => registered() === 0)).toBe(true);
63+
expect(registry().timer).toBe(null);
64+
});
65+
66+
test('closing a server closes connections to peers without the session layer', async () => {
67+
const { server, url } = await startServer();
68+
// a plain WebSocket never answers the session handshake, so it is never told __ts/bye
69+
const WebSocket = require('ws');
70+
const peer = new WebSocket(url);
71+
peer.on('error', () => {});
72+
await new Promise(resolve => peer.on('open', resolve));
73+
expect(await waitFor(() => registered() === 1)).toBe(true);
74+
75+
const peerClosed = new Promise(resolve => peer.on('close', resolve));
76+
server.close();
77+
await peerClosed;
78+
expect(await waitFor(() => registered() === 0)).toBe(true);
79+
expect(registry().timer).toBe(null);
80+
});
81+
82+
test('closing a server surfaces the close of a session held in grace', async () => {
83+
const server = new ToolSocket.Server({ port: 0, session: { graceMs: 60000 } }, 'server');
84+
running.push(server);
85+
await new Promise(resolve => server.server.on('listening', resolve));
86+
const url = `ws://127.0.0.1:${server.server.address().port}`;
87+
const connected = new Promise(resolve => server.addEventListener('connection', resolve));
88+
const client = new ToolSocket(new URL(url), 'testnet', 'web');
89+
running.push(client);
90+
const incoming = await connected;
91+
NB.enhance(incoming, { reconnect: false });
92+
await opened(client);
93+
expect(await waitFor(() => incoming.__ssn.peer === true)).toBe(true);
94+
expect(registered()).toBe(1);
95+
96+
// the transport drops without a toolsocket close(): the server holds the session for a
97+
// successor that, with the client's migration stopped, never comes
98+
client.__ssn.userClosed = true;
99+
client.socket.terminate();
100+
expect(await waitFor(() => incoming.__ssn.inGrace())).toBe(true);
101+
102+
const incomingClosed = new Promise(resolve => incoming.addEventListener('close', resolve));
103+
server.close();
104+
await incomingClosed;
105+
expect(registered()).toBe(0);
106+
});
107+
108+
test('a close listener that throws does not stop the server closing the rest', async () => {
109+
const server = new ToolSocket.Server({ port: 0, session: { graceMs: 60000 } }, 'server');
110+
running.push(server);
111+
await new Promise(resolve => server.server.on('listening', resolve));
112+
const url = `ws://127.0.0.1:${server.server.address().port}`;
113+
server.addEventListener('connection', socket => NB.enhance(socket, { reconnect: false }));
114+
const clients = [];
115+
for (let i = 0; i < 2; i++) {
116+
const client = new ToolSocket(new URL(url), 'testnet', 'web');
117+
running.push(client);
118+
clients.push(client);
119+
await opened(client);
120+
}
121+
expect(await waitFor(() => server.sockets.length === 2 && server.sockets.every(s => s.__ssn.peer === true))).toBe(true);
122+
expect(registered()).toBe(2);
123+
124+
// both sessions held in grace, so server.close() surfaces their closes, and so runs their
125+
// listeners, synchronously
126+
for (const client of clients) {
127+
client.__ssn.userClosed = true;
128+
client.socket.terminate();
129+
}
130+
expect(await waitFor(() => server.sockets.every(s => s.__ssn.inGrace()))).toBe(true);
131+
132+
const [first, second] = server.sockets;
133+
const listenerError = new Error('application close listener');
134+
first.addEventListener('close', () => {
135+
throw listenerError;
136+
});
137+
let secondClosed = false;
138+
second.addEventListener('close', () => {
139+
secondClosed = true;
140+
});
141+
142+
expect(() => server.close()).toThrow(listenerError);
143+
expect(secondClosed).toBe(true);
144+
expect(server.sockets.length).toBe(0);
145+
expect(registered()).toBe(0);
146+
expect(registry().timer).toBe(null);
147+
expect(server.server.address()).toBe(null); // stopped listening
148+
});
149+
150+
test('closing a reconnecting client releases it from the registry', async () => {
151+
const { url } = await startServer();
152+
const client = NB.connect(url, 'testnet', 'web', { reconnect: true });
153+
await opened(client);
154+
// once the session handshake is done, a close is sent as __ts/bye, so the server's end
155+
// closes at once rather than waiting out its grace for a successor
156+
expect(await waitFor(() => client.__ssn.peer === true)).toBe(true);
157+
expect(await waitFor(() => registered() === 2)).toBe(true);
158+
159+
const clientClosed = new Promise(resolve => client.addEventListener('close', resolve));
160+
client.close();
161+
await clientClosed;
162+
// the server's end goes when the client's close reaches it
163+
expect(await waitFor(() => registered() === 0)).toBe(true);
164+
expect(registry().timer).toBe(null);
165+
});

0 commit comments

Comments
 (0)