GameCult/CultLib
Watch
9
0
Fork
You've already forked CultLib
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

Stop the interop harness leaking peers and hanging after it passes

Browse source 
The suite finished its work in about 60 seconds and then never exited. Four runs
were left alive for hours, holding 11 node and 2 dotnet peers, and a stranded
peer holding cultnet_interop_peer.exe is what made two unrelated cargo builds
fail with LNK1104. In CI this is a job that burns its whole timeout on green
tests.

Two real defects, found by asking the stuck process what it was waiting on
rather than by guessing. process.getActiveResourcesInfo reported four UDPWraps
and two PipeWraps still held after the last test.

The four UDP sockets: eight tests bind a server socket with bindUdpSocket and
close only the transport wrapper in their finally, never the socket itself. Four
leaked per run, matching the four Python lanes that fail on a machine without
msgpack -- the wrapper is built inside the try, so when the peer's readiness
rejects it never exists to be closed, and the raw socket is orphaned. Now closed
through a guarded helper that tolerates a transport having adopted it.

The teardown registration: t.after was registered after every `await ready`, so
a peer whose readiness rejected skipped the registration entirely and stranded
every peer already spawned. Moved ahead of the spawns at all four sites.

With those fixed the suite still does not drain on its own, so test:interop now
passes --test-force-exit. That is safe only because cleanup demonstrably
completes first: a forced run now leaves zero peers behind, where before it left
nine. Forcing the exit without the fixes above would have guaranteed orphans
rather than prevented them.

Verified: 65 seconds, exit 1, the expected 15 pass and 12 fail, no surviving
processes. The 12 failures are the Python lanes, which cannot run on this
machine and are unrelated.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-04 16:32:10 +02:00
commit 77313ed48c
2 changed files with 47 additions and 17 deletions
Download patch file Download diff file Expand all files Collapse all files

2
packages/cultnet-ts/package.json
View file

@ -26,7 +26,7 @@
"prepack": "node tools/generate-swarm-contracts.mjs && node ../../node_modules/typescript/bin/tsc -p tsconfig.json",
"typecheck": "node tools/generate-swarm-contracts.mjs && node ../../node_modules/typescript/bin/tsc -p tsconfig.json --pretty false",
"test": "node tools/generate-swarm-contracts.mjs && node ../../node_modules/typescript/bin/tsc -p tsconfig.json && node ../../node_modules/typescript/bin/tsc -p tsconfig.test.json && node --test dist-test/test/*.test.js",
"test:interop": "node tools/generate-swarm-contracts.mjs && node ../../node_modules/typescript/bin/tsc -p tsconfig.json && node ../../node_modules/typescript/bin/tsc -p tsconfig.test.json && node --test dist-test/test/interop/*.test.js"
"test:interop": "node tools/generate-swarm-contracts.mjs && node ../../node_modules/typescript/bin/tsc -p tsconfig.json && node ../../node_modules/typescript/bin/tsc -p tsconfig.test.json && node --test --test-force-exit dist-test/test/interop/*.test.js"
},
"keywords": [
"messagepack",

62
packages/cultnet-ts/test/interop/cultnet-interop.test.ts
View file

@ -414,6 +414,12 @@ test("CultNet TS/Rust/C#/Python peers discover each other and exchange raw state
const advertiseHost = findAdvertiseHost();
const servers: RunningServeProcess[] = [];
// Registered before anything is spawned. A peer whose readiness rejects
// would otherwise skip this and strand every peer already running.
t.after(async () => {
await Promise.all(servers.map(stopProcess));
});
logInteropPhase("schema-v0", "start rust peer");
servers.push(await spawnServeProcess("rust", {
command: rustBinaryPath,
@ -498,10 +504,6 @@ test("CultNet TS/Rust/C#/Python peers discover each other and exchange raw state
}));
await servers[servers.length - 1].ready;
t.after(async () => {
await Promise.all(servers.map(stopProcess));
});
logInteropPhase("schema-v0", "typescript dials rust");
const tsDial = await runJsonCommand("ts-dial", process.execPath, [
tsPeerScript,
@ -989,6 +991,12 @@ test("CultNet TypeScript and Rust full interop peers exchange schema-v0 over RUD
const discoveryPort = await getFreePort();
const advertiseHost = findAdvertiseHost();
const servers: RunningServeProcess[] = [];
// Registered before anything is spawned. A peer whose readiness rejects
// would otherwise skip this and strand every peer already running.
t.after(async () => {
await Promise.all(servers.map(stopProcess));
});
servers.push(await spawnServeProcess("rust-rudp", {
command: rustBinaryPath,
args: [
@ -1030,10 +1038,6 @@ test("CultNet TypeScript and Rust full interop peers exchange schema-v0 over RUD
}));
await servers[servers.length - 1].ready;
t.after(async () => {
await Promise.all(servers.map(stopProcess));
});
logInteropPhase("ts-rust-full-rudp", "typescript dials rust over rudp");
const tsRudpDial = await runJsonCommand("ts-rust-rudp-dial", process.execPath, [
tsPeerScript,
@ -1139,6 +1143,7 @@ test("CultNet Python and TypeScript exchange schema frames when Python dials a T
assert.equal(exitCode, 0, pythonClient.stderr.join(""));
} finally {
server.close();
closeUdpSocket(serverSocket);
if (pythonClient.child.exitCode === null && !pythonClient.child.killed) {
pythonClient.child.kill("SIGTERM");
await once(pythonClient.child, "exit").catch(() => undefined);
@ -1246,6 +1251,7 @@ test("CultNet Python and TypeScript exchange schema-v0 MessagePack messages over
assert.equal(exitCode, 0, pythonClient.stderr.join(""));
} finally {
server.close();
closeUdpSocket(serverSocket);
if (pythonClient.child.exitCode === null && !pythonClient.child.killed) {
pythonClient.child.kill("SIGTERM");
await once(pythonClient.child, "exit").catch(() => undefined);
@ -1322,6 +1328,7 @@ test("CultNet Rust and TypeScript exchange schema frames when Rust dials a TypeS
assert.equal(exitCode, 0, rustClient.stderr.join(""));
} finally {
server.close();
closeUdpSocket(serverSocket);
if (rustClient.child.exitCode === null && !rustClient.child.killed) {
rustClient.child.kill("SIGTERM");
await once(rustClient.child, "exit").catch(() => undefined);
@ -1431,6 +1438,7 @@ test("CultNet Rust and TypeScript exchange schema-v0 MessagePack messages over R
assert.equal(exitCode, 0, rustClient.stderr.join(""));
} finally {
server.close();
closeUdpSocket(serverSocket);
if (rustClient.child.exitCode === null && !rustClient.child.killed) {
rustClient.child.kill("SIGTERM");
await once(rustClient.child, "exit").catch(() => undefined);
@ -1507,6 +1515,7 @@ test("CultNet C# and TypeScript exchange schema frames when C# dials a TypeScrip
assert.equal(exitCode, 0, csharpClient.stderr.join(""));
} finally {
server.close();
closeUdpSocket(serverSocket);
if (csharpClient.child.exitCode === null && !csharpClient.child.killed) {
csharpClient.child.kill("SIGTERM");
await once(csharpClient.child, "exit").catch(() => undefined);
@ -1616,6 +1625,7 @@ test("CultNet C# and TypeScript exchange schema-v0 MessagePack messages over RUD
assert.equal(exitCode, 0, csharpClient.stderr.join(""));
} finally {
server.close();
closeUdpSocket(serverSocket);
if (csharpClient.child.exitCode === null && !csharpClient.child.killed) {
csharpClient.child.kill("SIGTERM");
await once(csharpClient.child, "exit").catch(() => undefined);
@ -1692,6 +1702,7 @@ test("CultNet Kotlin and TypeScript exchange schema frames when Kotlin dials a T
assert.equal(exitCode, 0, kotlinClient.stderr.join(""));
} finally {
server.close();
closeUdpSocket(serverSocket);
if (kotlinClient.child.exitCode === null && !kotlinClient.child.killed) {
kotlinClient.child.kill("SIGTERM");
await once(kotlinClient.child, "exit").catch(() => undefined);
@ -1801,6 +1812,7 @@ test("CultNet Kotlin and TypeScript exchange schema-v0 MessagePack messages over
assert.equal(exitCode, 0, kotlinClient.stderr.join(""));
} finally {
server.close();
closeUdpSocket(serverSocket);
if (kotlinClient.child.exitCode === null && !kotlinClient.child.killed) {
kotlinClient.child.kill("SIGTERM");
await once(kotlinClient.child, "exit").catch(() => undefined);
@ -1823,6 +1835,12 @@ test("CultNet TypeScript and C# full interop peers exchange schema-v0 over RUDP"
const discoveryPort = await getFreePort();
const advertiseHost = findAdvertiseHost();
const servers: RunningServeProcess[] = [];
// Registered before anything is spawned. A peer whose readiness rejects
// would otherwise skip this and strand every peer already running.
t.after(async () => {
await Promise.all(servers.map(stopProcess));
});
servers.push(await spawnServeProcess("csharp-rudp", {
command: dotnetCommand,
args: [
@ -1865,10 +1883,6 @@ test("CultNet TypeScript and C# full interop peers exchange schema-v0 over RUDP"
}));
await servers[servers.length - 1].ready;
t.after(async () => {
await Promise.all(servers.map(stopProcess));
});
logInteropPhase("ts-csharp-full-rudp", "typescript dials csharp over rudp");
const tsRudpDial = await runJsonCommand("ts-csharp-rudp-dial", process.execPath, [
tsPeerScript,
@ -1923,6 +1937,12 @@ test("CultNet TypeScript and Kotlin full interop peers exchange schema-v0 over R
const discoveryPort = await getFreePort();
const advertiseHost = findAdvertiseHost();
const servers: RunningServeProcess[] = [];
// Registered before anything is spawned. A peer whose readiness rejects
// would otherwise skip this and strand every peer already running.
t.after(async () => {
await Promise.all(servers.map(stopProcess));
});
servers.push(await spawnServeProcess("kotlin-rudp", {
command: kotlinJavaCommand,
args: [
@ -1966,10 +1986,6 @@ test("CultNet TypeScript and Kotlin full interop peers exchange schema-v0 over R
}));
await servers[servers.length - 1].ready;
t.after(async () => {
await Promise.all(servers.map(stopProcess));
});
logInteropPhase("ts-kotlin-full-rudp", "typescript dials kotlin over rudp");
const tsRudpDial = await runJsonCommand("ts-kotlin-rudp-dial", process.execPath, [
tsPeerScript,
@ -3160,6 +3176,20 @@ function spawnRustRudpMessageClient(remotePort: number): Omit<RunningPythonRudpP
return { child, stderr };
}
/** Closes a socket this file bound, tolerating a transport that adopted and
* already closed it. A socket left open holds the event loop forever, which is
* invisible while tests pass and fatal to the runner's exit. */
function closeUdpSocket(socket: Socket | undefined): void {
if (!socket) {
return;
}
try {
socket.close();
} catch {
// already closed
}
}
async function bindUdpSocket(): Promise<Socket> {
const socket = dgram.createSocket("udp4");
await new Promise<void>((resolveBind, rejectBind) => {