Fix an orphaned pendingSends race, add Cut 5 runtime-release and eviction tests
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 <noreply@anthropic.com>
This commit is contained in:
parent
992d82e095
commit
e943159a47
2 changed files with 116 additions and 0 deletions
|
|
@ -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<void> {
|
||||
return new Promise<void>((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);
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
},
|
||||
);
|
||||
|
|
|
|||
Loading…
Reference in a new issue