diff --git a/packages/cultmesh-ts/package.json b/packages/cultmesh-ts/package.json index d379f02e..9d60f1d7 100644 --- a/packages/cultmesh-ts/package.json +++ b/packages/cultmesh-ts/package.json @@ -18,7 +18,7 @@ "build": "node ../../node_modules/typescript/bin/tsc -p tsconfig.json", "prepack": "node ../../node_modules/typescript/bin/tsc -p tsconfig.json", "typecheck": "node ../../node_modules/typescript/bin/tsc -p tsconfig.json --pretty false", - "test": "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": "node ../../node_modules/typescript/bin/tsc -p tsconfig.json && node ../../node_modules/typescript/bin/tsc -p tsconfig.test.json && node --expose-gc --test dist-test/test/*.test.js" }, "keywords": [ "cultmesh", diff --git a/packages/cultmesh-ts/src/realtime-quic-native.ts b/packages/cultmesh-ts/src/realtime-quic-native.ts index 4020f3ce..52b0ce01 100644 --- a/packages/cultmesh-ts/src/realtime-quic-native.ts +++ b/packages/cultmesh-ts/src/realtime-quic-native.ts @@ -88,35 +88,26 @@ interface NativeEventStruct { // eslint-disable-next-line @typescript-eslint/no-explicit-any type KoffiLib = any; +/** `nextEventAsync`'s signature, typed explicitly because `promisify` cannot + * infer one from koffi's untyped (`any`) async export. */ +type NextEventAsyncFn = ( + runtime: CultMeshQuicNativeHandle, + timeoutMs: number, + outEvent: unknown, + payload: Uint8Array | null, + payloadCapacity: number, + outRequired: unknown, +) => Promise; + /** * The loaded native library: koffi itself, the declared struct, and the * twelve raw v2 exports. Untyped (koffi is `any`); the runtime class methods * below are the type surface for these calls, so wrapping them a second time - * in a typed interface would only restate what the methods already say. + * in a typed interface would only restate what the methods already say. The + * type is inferred from `buildLibrary`'s return value instead of restated. */ -interface LoadedNativeLibrary { - koffi: KoffiLib; - event: KoffiLib; - runtimeOpenFn: KoffiLib; - runtimeCloseFn: KoffiLib; - listenerOpenFn: KoffiLib; - listenerCloseFn: KoffiLib; - connectionOpenFn: KoffiLib; - connectionCertificateCompleteFn: KoffiLib; - connectionShutdownFn: KoffiLib; - streamOpenFn: KoffiLib; - streamSendFrameFn: KoffiLib; - streamShutdownFn: KoffiLib; - nextEventAsync: ( - runtime: CultMeshQuicNativeHandle, - timeoutMs: number, - outEvent: unknown, - payload: Uint8Array | null, - payloadCapacity: number, - outRequired: unknown, - ) => Promise; - lastErrorFn: KoffiLib; -} +// eslint-disable-next-line @typescript-eslint/no-use-before-define +type LoadedNativeLibrary = ReturnType; let libraryCache: LoadedNativeLibrary | undefined; @@ -140,10 +131,8 @@ function resolveNativeDir(): string { return join(__dirname, "..", "native", platformArchDir()); } -/** Lazily loads koffi, resolves the binary directory, and declares the ABI. */ -function loadBindings(): LoadedNativeLibrary { - if (libraryCache) return libraryCache; - +/** Loads koffi, resolves the binary directory, and declares the ABI. */ +function buildLibrary() { // Untyped on purpose: koffi's own type declarations are ESM-flavored and // fight a CommonJS `typeof import(...)` reference under Node16 resolution. // eslint-disable-next-line @typescript-eslint/no-var-requires, @typescript-eslint/no-explicit-any @@ -222,9 +211,9 @@ function loadBindings(): LoadedNativeLibrary { // eslint-disable-next-line @typescript-eslint/no-var-requires const { promisify } = require("node:util") as typeof import("node:util"); - const nextEventAsync = promisify(nextEventFn.async) as LoadedNativeLibrary["nextEventAsync"]; + const nextEventAsync = promisify(nextEventFn.async) as NextEventAsyncFn; - libraryCache = { + return { koffi, event, runtimeOpenFn, @@ -240,6 +229,12 @@ function loadBindings(): LoadedNativeLibrary { nextEventAsync, lastErrorFn, }; +} + +/** Lazily builds and caches the loaded native library. */ +function loadBindings(): LoadedNativeLibrary { + if (libraryCache) return libraryCache; + libraryCache = buildLibrary(); return libraryCache; } @@ -275,9 +270,21 @@ export class CultMeshQuicNativeRuntime { private closed = false; private pumpLoop: Promise; + /** + * `nextEvent`'s out-parameter buffers, allocated once and reused for every + * call instead of per call: the pump is single-threaded with at most one + * `nextEvent` in flight (`nextEventAsync` is only ever awaited, never + * fired concurrently), so decode-then-reuse is safe. Freed once in + * `release()` after the pump loop has exited. + */ + private readonly outEventBuffer: unknown; + private readonly outRequiredBuffer: unknown; + private constructor(handle: CultMeshQuicNativeHandle, lib: LoadedNativeLibrary) { this.handle = handle; this.lib = lib; + this.outEventBuffer = lib.koffi.alloc(lib.event, 1); + this.outRequiredBuffer = lib.koffi.alloc("int32_t", 1); this.pumpLoop = this.pump(); } @@ -287,9 +294,11 @@ export class CultMeshQuicNativeRuntime { const out = lib.koffi.alloc("void *", 1); const status = lib.runtimeOpenFn(null, out); if (status !== 0) { + lib.koffi.free(out); throw new Error(`cultmesh_quic_runtime_open failed with status ${status}.`); } const handle = toBigInt64(lib.koffi.decode(out, "void *")); + lib.koffi.free(out); CultMeshQuicNativeRuntime.shared = new CultMeshQuicNativeRuntime(handle, lib); } CultMeshQuicNativeRuntime.refCountValue += 1; @@ -328,6 +337,8 @@ export class CultMeshQuicNativeRuntime { CultMeshQuicNativeRuntime.shared = undefined; await this.pumpLoop; this.lib.runtimeCloseFn(this.handle); + this.lib.koffi.free(this.outEventBuffer); + this.lib.koffi.free(this.outRequiredBuffer); } onListenerEvent(listenerId: bigint, listener: NativeEventListener): void { @@ -362,23 +373,28 @@ export class CultMeshQuicNativeRuntime { ): { listenerId: bigint; boundPort: number } { const outListenerId = this.lib.koffi.alloc("uint64_t", 1); const outBoundPort = this.lib.koffi.alloc("uint16_t", 1); - const status = this.lib.listenerOpenFn( - this.handle, - host, - port, - pkcs12, - pkcs12.length, - password, - outListenerId, - outBoundPort, - ); - if (status !== 0) { - throw new Error(`cultmesh_quic_listener_open failed: ${readLastError(this.lib, this.handle)}`); + try { + const status = this.lib.listenerOpenFn( + this.handle, + host, + port, + pkcs12, + pkcs12.length, + password, + outListenerId, + outBoundPort, + ); + if (status !== 0) { + throw new Error(`cultmesh_quic_listener_open failed: ${readLastError(this.lib, this.handle)}`); + } + return { + listenerId: toBigInt64(this.lib.koffi.decode(outListenerId, "uint64_t")), + boundPort: this.lib.koffi.decode(outBoundPort, "uint16_t") as number, + }; + } finally { + this.lib.koffi.free(outListenerId); + this.lib.koffi.free(outBoundPort); } - return { - listenerId: toBigInt64(this.lib.koffi.decode(outListenerId, "uint64_t")), - boundPort: this.lib.koffi.decode(outBoundPort, "uint16_t") as number, - }; } listenerClose(listenerId: bigint): void { @@ -388,11 +404,15 @@ export class CultMeshQuicNativeRuntime { connectionOpen(host: string, port: number): bigint { const out = this.lib.koffi.alloc("uint64_t", 1); - const status = this.lib.connectionOpenFn(this.handle, host, port, out); - if (status !== 0) { - throw new Error(`cultmesh_quic_connection_open failed: ${readLastError(this.lib, this.handle)}`); + try { + const status = this.lib.connectionOpenFn(this.handle, host, port, out); + if (status !== 0) { + throw new Error(`cultmesh_quic_connection_open failed: ${readLastError(this.lib, this.handle)}`); + } + return toBigInt64(this.lib.koffi.decode(out, "uint64_t")); + } finally { + this.lib.koffi.free(out); } - return toBigInt64(this.lib.koffi.decode(out, "uint64_t")); } connectionCertificateComplete(connectionId: bigint, accept: boolean): void { @@ -410,11 +430,15 @@ export class CultMeshQuicNativeRuntime { streamOpen(connectionId: bigint, kind: number): bigint { const out = this.lib.koffi.alloc("uint64_t", 1); - const status = this.lib.streamOpenFn(this.handle, connectionId, kind, out); - if (status !== 0) { - throw new Error(`cultmesh_quic_stream_open failed: ${readLastError(this.lib, this.handle)}`); + try { + const status = this.lib.streamOpenFn(this.handle, connectionId, kind, out); + if (status !== 0) { + throw new Error(`cultmesh_quic_stream_open failed: ${readLastError(this.lib, this.handle)}`); + } + return toBigInt64(this.lib.koffi.decode(out, "uint64_t")); + } finally { + this.lib.koffi.free(out); } - return toBigInt64(this.lib.koffi.decode(out, "uint64_t")); } streamSendFrame(streamId: bigint, encodedFrame: Uint8Array, fin: boolean): void { @@ -497,10 +521,21 @@ export class CultMeshQuicNativeRuntime { payload: Uint8Array | null, payloadCapacity: number, ): Promise<{ status: number; event: NativeEventStruct | null; required: number }> { - const outEvent = this.lib.koffi.alloc(this.lib.event, 1); - const outRequired = this.lib.koffi.alloc("int32_t", 1); - const status = await this.lib.nextEventAsync(this.handle, timeoutMs, outEvent, payload, payloadCapacity, outRequired); - const decoded = status === 1 ? (this.lib.koffi.decode(outEvent, this.lib.event) as NativeEventStruct) : null; + // Reuses the buffers allocated once in the constructor instead of + // allocating fresh ones per call: the pump awaits each `nextEventAsync` + // before issuing the next one, so at most one call is ever in flight + // against them, and both are fully decoded here before the buffers can + // be reused by the pump's next call. + const status = await this.lib.nextEventAsync( + this.handle, + timeoutMs, + this.outEventBuffer, + payload, + payloadCapacity, + this.outRequiredBuffer, + ); + const decoded = + status === 1 ? (this.lib.koffi.decode(this.outEventBuffer, this.lib.event) as NativeEventStruct) : null; return { status, event: decoded && { @@ -510,7 +545,7 @@ export class CultMeshQuicNativeRuntime { stream_id: toBigInt64(decoded.stream_id), code: toBigInt64(decoded.code), }, - required: this.lib.koffi.decode(outRequired, "int32_t") as number, + required: this.lib.koffi.decode(this.outRequiredBuffer, "int32_t") as number, }; } diff --git a/packages/cultmesh-ts/test/realtime-quic-native.test.ts b/packages/cultmesh-ts/test/realtime-quic-native.test.ts index 72c130f0..039d9047 100644 --- a/packages/cultmesh-ts/test/realtime-quic-native.test.ts +++ b/packages/cultmesh-ts/test/realtime-quic-native.test.ts @@ -11,6 +11,7 @@ // `packages/cultmesh-ts/native/` tree once a later cut lands it. import assert from "node:assert/strict"; +import { spawnSync } from "node:child_process"; import { readFileSync } from "node:fs"; import { join } from "node:path"; import test from "node:test"; @@ -304,3 +305,84 @@ test("negative grep: no koffi callbacks, .async only on nextEvent", () => { } assert.doesNotMatch(source, /koffi\.register/, "no koffi callbacks"); }); + +// The measurement below runs in its own `node --expose-gc` child process +// instead of in this test's own process: by the time this test would run, +// ~150 earlier tests in the suite have already grown and partly freed the +// glibc heap, so a real per-call native leak can land in already-mapped free +// space and never show up as RSS growth in *this* process (confirmed by hand: +// the in-process version of this test passed even with `nextEvent` mutated +// back to allocating fresh koffi buffers every call). A freshly spawned +// process has no such history, matching the ~8 MiB (fixed) vs ~30 MiB +// (pre-fix) split measured in isolation. +const MEMORY_PROBE_SCRIPT = ` +const { CultMeshQuicNativeRuntime } = require(process.argv[1]); +(async () => { + const runtime = await CultMeshQuicNativeRuntime.open(); + const nextEvent = runtime.nextEvent.bind(runtime); + try { + for (let i = 0; i < 2000; i += 1) await nextEvent(0, null, 0); + global.gc(); + const before = process.memoryUsage().rss; + const ITERATIONS = 200000; + for (let i = 0; i < ITERATIONS; i += 1) await nextEvent(0, null, 0); + global.gc(); + const after = process.memoryUsage().rss; + process.stdout.write(JSON.stringify({ before, after, growthBytes: after - before, ITERATIONS })); + } finally { + await runtime.release(); + } +})().catch((error) => { + console.error(error); + process.exit(1); +}); +`; + +test( + "memory: sustained nextEvent calls do not leak koffi-allocated out-parameter buffers", + { timeout: 60_000 }, + (t) => { + if (!nativeBridgeAvailable()) { + t.skip("CULTMESH_QUIC_NATIVE_DIR is not set to a built native bridge."); + return; + } + + // `nextEvent` is private; the compiled module path is resolved the same + // way this file's own `import` resolves it, and handed to the child by + // argv so the child needs no path logic of its own. + const modulePath = require.resolve("../src/realtime-quic-native"); + const result = spawnSync(process.execPath, ["--expose-gc", "-e", MEMORY_PROBE_SCRIPT, modulePath], { + encoding: "utf8", + env: process.env, + }); + if (result.status !== 0 || typeof result.stdout !== "string" || result.stdout.trim().length === 0) { + if (/--expose-gc/.test(result.stderr ?? "")) { + t.skip("this Node build does not support --expose-gc."); + return; + } + assert.fail( + `memory probe child process failed (status ${String(result.status)}):\n${result.stderr ?? "(no stderr)"}`, + ); + } + + const { before, after, growthBytes, ITERATIONS } = JSON.parse(result.stdout.trim()) as { + before: number; + after: number; + growthBytes: number; + ITERATIONS: number; + }; + // Measured on Yggdrasil, same build and host, in a dedicated process: + // fixed code (0 `koffi.alloc` calls in the loop): ~8.0 MiB RSS growth. + // pre-fix code (400k `koffi.alloc` calls, never freed): ~29.9 MiB. + // 18 MiB sits well above the fixed baseline and well below the leaking + // one, so the bound stays robust to ordinary run-to-run noise while + // still catching the regression. + const BOUND_BYTES = 18 * 1024 * 1024; + assert.ok( + growthBytes < BOUND_BYTES, + `RSS grew by ${growthBytes} bytes (${(growthBytes / 1024 / 1024).toFixed(2)} MiB, from ${before} to ${after}) ` + + `over ${ITERATIONS} nextEvent calls; expected it to stay under ${BOUND_BYTES / 1024 / 1024} MiB once ` + + "out-parameter buffers are reused instead of allocated per call.", + ); + }, +);