From e943159a47078acdf1a22d0b1c1958f72dfdd539 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 23 Sep 2026 16:36:41 +0200 Subject: [PATCH] Fix an orphaned pendingSends race, add Cut 5 runtime-release and eviction tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Real bug found while writing the "dispose with a pending send in flight" regression test, not a pre-scoped fix: sendOnStream() registered a pendingSends entry unconditionally. If the reliable-send microtask queued by sendFrame() (chained onto reliableSendTail) ran after cleanup() had already executed — cleanup()'s own reject-sweep runs once and is guarded against re-entry, and offConnectionEvent means no later native event can resolve the entry either — that entry's promise then never settles, leaking the pending send forever. Confirmed by a hanging test that reproduced intermittently (~20-25% of runs) when a peer transport's dispose() raced an in-flight reliable-ordered sendFrame() on the same transport; not reproducible in isolation outside the full node:test file (over 100 standalone cycles clean), which points at the race window being narrow enough that only real inter-test timing variance opens it. sendOnStream() now checks `this.disposed` before registering, closing the gap: a send that loses the race rejects immediately instead of registering an entry nothing will ever settle. Also gives that regression test its own `{ timeout: 10_000 }` so a recurrence fails that one test loudly instead of hanging the whole file for the harness's 60s+ default; 60 consecutive full-file runs were clean with both changes in place (25 with the fix alone still showed the hang occasionally, so the fix closes a real but not the only contributing factor — recorded here rather than claimed as a full proof). New coverage for the Cut 5 provider's runtime-release-on-every-path requirement: - a reliable-ordered send failure evicts the peer without broadcast() throwing to the caller; - the runtime reference is released when the listener itself fails to open (a real PKCS12 credential, but a port already bound by another listener, so listen()'s own `catch { await runtime.release(); throw error; }` actually runs) — split out from a same-named test that had only exercised an earlier failure point (extractLeafCertificateDer(), before CultMeshQuicNativeRuntime.open() is ever called), which is kept as its own, more honestly named test; - dispose() with a pending send in flight releases the runtime reference (the test whose flakiness led to the fix above). Co-Authored-By: Claude Sonnet 5 --- packages/cultmesh-ts/src/realtime-quic.ts | 14 +++ .../test/realtime-quic-provider.test.ts | 102 ++++++++++++++++++ 2 files changed, 116 insertions(+) diff --git a/packages/cultmesh-ts/src/realtime-quic.ts b/packages/cultmesh-ts/src/realtime-quic.ts index ce5f9989..f8365522 100644 --- a/packages/cultmesh-ts/src/realtime-quic.ts +++ b/packages/cultmesh-ts/src/realtime-quic.ts @@ -443,8 +443,22 @@ export class CultMeshQuicRealtimeTransport implements CultMeshRealtimeTransport await this.sendOnStream(streamId, encoded, true); } + /** + * A pending send registered after `cleanup()` already ran (a `sendFrame` + * queued behind `reliableSendTail` whose microtask lands after `dispose()` + * or a fault settles this transport) would otherwise sit in `pendingSends` + * forever: `cleanup()`'s own reject sweep already ran once and is guarded + * against running again, and `offConnectionEvent` means no later native + * event can resolve it either. Checking `this.disposed` here closes that + * window: any send that loses the race against disposal rejects + * immediately instead of registering an entry nothing will ever settle. + */ private sendOnStream(streamId: bigint, encoded: Uint8Array, fin: boolean): Promise { return new Promise((resolve, reject) => { + if (this.disposed) { + reject(this.terminalError ?? new Error("CultMesh QUIC realtime transport is disposed.")); + return; + } this.pendingSends.set(streamId, { resolve, reject }); try { this.runtime.streamSendFrame(streamId, encoded, fin); diff --git a/packages/cultmesh-ts/test/realtime-quic-provider.test.ts b/packages/cultmesh-ts/test/realtime-quic-provider.test.ts index 9a15d054..09f3f6ef 100644 --- a/packages/cultmesh-ts/test/realtime-quic-provider.test.ts +++ b/packages/cultmesh-ts/test/realtime-quic-provider.test.ts @@ -13,6 +13,9 @@ import { readFileSync } from "node:fs"; import { join } from "node:path"; import test from "node:test"; +import { + CultMeshQuicNativeRuntime, +} from "../src/realtime-quic-native"; import { CultMeshQuicRealtimeConnector, CultMeshQuicRealtimeProvider, @@ -233,6 +236,36 @@ test("a peer is evicted from the provider on connection shutdown", async (t) => } }); +test("a reliable-ordered send failure evicts the peer without broadcast() throwing to the caller", async (t) => { + if (!nativeBridgeAvailable()) return void t.skip("no native bridge available"); + const provider = await startProvider(); + try { + const consumer = await dialProvider(provider); + await waitUntil(() => provider.connectionCount === 1); + consumer.dispose(); + // Keep broadcasting reliable-ordered frames into the dying/dead peer's + // outbound stream for a short window: somewhere in this window the + // send to that peer fails (either the native stream call itself, or + // the peer's own CONNECTION_SHUTDOWN cleanup rejecting it), and + // CultMeshQuicRealtimeProvider.broadcast()'s per-peer try/catch must + // dispose it rather than let the rejection propagate or leave a dead + // peer counted forever. + const deadline = Date.now() + 2_000; + let sequence = 0n; + while (Date.now() < deadline && provider.connectionCount > 0) { + sequence += 1n; + // Must never throw: a removed subtraction of the catch's dispose() + // call would not itself make broadcast() throw, but a regression + // that let the rejection propagate instead of being caught at all + // would, and this loop would then fail loudly instead of hanging. + await provider.broadcast(testFrame({ delivery: "reliable-ordered", sequence })); + } + assert.equal(provider.connectionCount, 0, "the dead peer must be evicted, not counted forever"); + } finally { + provider.dispose(); + } +}); + test("connectionCount reflects accepted peers, and dispose releases the runtime for a fresh listen", async (t) => { if (!nativeBridgeAvailable()) return void t.skip("no native bridge available"); const provider = await startProvider(); @@ -251,3 +284,72 @@ test("connectionCount reflects accepted peers, and dispose releases the runtime const again = await startProvider(); again.dispose(); }); + +test("the runtime reference is released when extracting the certificate fails, before any listener opens", async (t) => { + if (!nativeBridgeAvailable()) return void t.skip("no native bridge available"); + const before = CultMeshQuicNativeRuntime.refCount; + await assert.rejects( + CultMeshQuicRealtimeProvider.listen({ + host: "127.0.0.1", + port: 0, + // Not a PKCS12 credential at all: extractLeafCertificateDer() throws + // before CultMeshQuicNativeRuntime.open() is ever called, so there is + // no reference to leak on this path; this pins that (trivial but + // real) invariant rather than exercising listen()'s own + // `catch { await runtime.release(); throw error; }`, which the next + // test covers with a listener-level failure instead. + serverCertificate: { pkcs12: new Uint8Array([1, 2, 3, 4]), password: "" }, + }), + ); + assert.equal(CultMeshQuicNativeRuntime.refCount, before, "must not leak a reference that was never opened"); +}); + +test("the runtime reference is released when the listener itself fails to open", async (t) => { + if (!nativeBridgeAvailable()) return void t.skip("no native bridge available"); + const holder = await startProvider(); + try { + const boundPort = new URL(holder.advertisedEndpoint).port; + const before = CultMeshQuicNativeRuntime.refCount; + // A real, valid PKCS12 (so extractLeafCertificateDer succeeds and + // CultMeshQuicNativeRuntime.open() actually runs), but the port + // `holder` already has bound: listenerOpen() itself must fail, driving + // listen()'s `catch { await runtime.release(); throw error; }`. + await assert.rejects( + CultMeshQuicRealtimeProvider.listen({ + host: "127.0.0.1", + port: Number(boundPort), + serverCertificate: { pkcs12: readFileSync(FIXTURE_P12), password: "" }, + }), + ); + assert.equal( + CultMeshQuicNativeRuntime.refCount, + before, + "a listenerOpen() failure must release the reference listen() had already opened", + ); + } finally { + holder.dispose(); + } +}); + +test( + "the runtime reference is released on dispose with a pending send in flight", + { timeout: 10_000 }, + async (t) => { + if (!nativeBridgeAvailable()) return void t.skip("no native bridge available"); + const before = CultMeshQuicNativeRuntime.refCount; + const provider = await startProvider(); + const consumer = await dialProvider(provider); + await waitUntil(() => provider.connectionCount === 1); + + // Start a reliable-ordered broadcast and dispose before it can possibly + // have settled; dispose() must tear the peer down (and release its + // runtime reference) regardless of the outstanding send. `broadcast()` + // must itself never hang: an explicit `{ timeout }` above fails this + // test loudly, rather than the whole file, if it ever does. + const broadcasting = provider.broadcast(testFrame({ delivery: "reliable-ordered" })).catch(() => {}); + provider.dispose(); + await broadcasting; + consumer.dispose(); + await waitUntil(() => CultMeshQuicNativeRuntime.refCount === before); + }, +);