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

Add tests pinning fixes 1-6, M2, M3, M6-M9, and dedupe the two nativeBridgeAvailable helpers

Browse source 
Each rule below fails under its own mutation of the code fixed in the previous
two commits; verified by hand and restored, per the fix-batch brief, not
committed as a mutation harness (the operator retired that pattern).

- realtime-quic-native.test.ts: release() refuses a call with no outstanding
  reference (fix 2).
- realtime-quic-consumer.test.ts: a stream-kind/delivery mismatch faults only
  its connection while a second, independent connection through the same
  pump keeps working (fix 1 / M4); a peer-initiated shutdown rejects a
  pending receive and cleans up exactly once, so a further dispose is a
  no-op and a fresh receive rejects instead of hanging (fix 3); a slow
  latest-only reader accumulates at most one pending frame per key, and an
  equal generation is dropped outright (fix 6, M10); a real connected
  transport's isVerifiedFor rejects a generation other than the one it
  connected with, and the session manager disposes a real (not stubbed)
  transport whose generation no longer matches a freshly verified route
  (M3); the advertised certificate pin matches case-insensitively (M2).
- realtime-quic-session-manager.test.ts (new): eviction of a dead cached
  session and redial on the next connect, single-flight sharing one dial
  across concurrent callers, refusing a transport that lands after dispose()
  (all fix 4); the race returning on first success without waiting out a
  slow candidate (fix 5); the maxRacedCandidates cap (M6); the protocolIds
  and authorityRuntimeId route filters (M7, M8); the session key including
  authorityRuntimeId (M9). Trust verification is skipped via
  mode: "local-development" over loopback endpoints, since these tests are
  about the session manager's own binding/racing/caching rules, not route
  trust (already pinned by realtime-quic-consumer.test.ts's trust
  negatives).

Subtraction: `nativeBridgeAvailable` was defined twice, disagreeing on
whether to check the dependency library too; both test files now import one
definition from test/support/native-bridge.ts (the stricter of the two).

Gap, reported rather than papered over: the stream-shutdown pruning half of
fix 6 (streamKinds) has no direct test, because the map is private and
nothing here exposes it; the committed test only shows a second frame still
arrives after the pruned stream's shutdown, which does not by itself prove
the entry was removed.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-23 14:38:46 +02:00
commit 047c55c453
4 changed files with 566 additions and 26 deletions
Download patch file Download diff file Expand all files Collapse all files

286
packages/cultmesh-ts/test/realtime-quic-consumer.test.ts
View file

@ -26,6 +26,7 @@ import {
CultMeshQuicNativeRuntime,
CULTMESH_QUIC_EVENT_LISTENER_NEW_CONNECTION,
CULTMESH_QUIC_STREAM_LATEST_ONLY,
CULTMESH_QUIC_STREAM_RELIABLE,
} from "../src/realtime-quic-native";
import {
CultMeshQuicRealtimeConnector,
@ -38,6 +39,7 @@ import {
type CultMeshRealtimeTransportConnector,
} from "../src/realtime-quic";
import { encodeRealtimeFrame, type CultMeshRealtimeFrame } from "../src/realtime-wire";
import { nativeBridgeAvailable } from "./support/native-bridge";
// __dirname at runtime is dist-test/test; fixtures are binary and are never
// compiled/copied there, so they are read from their source location, two
@ -45,18 +47,6 @@ import { encodeRealtimeFrame, type CultMeshRealtimeFrame } from "../src/realtime
const FIXTURE_P12 = join(__dirname, "..", "..", "test", "fixtures", "quic-test.p12");
const FIXTURE_DER = join(__dirname, "..", "..", "test", "fixtures", "quic-test-cert.der");
function nativeBridgeAvailable(): boolean {
const dir = process.env.CULTMESH_QUIC_NATIVE_DIR;
if (!dir) return false;
try {
const bridge = process.platform === "win32" ? "gamecult_mesh_quic_native.dll" : "libgamecult_mesh_quic_native.so";
readFileSync(join(dir, bridge));
return true;
} catch {
return false;
}
}
function fixturePinHex(): string {
return createHash("sha256").update(readFileSync(FIXTURE_DER)).digest("hex");
}
@ -100,6 +90,17 @@ class RawQuicListener {
this.runtime.streamSendFrame(streamId, encodeRealtimeFrame(frame), true);
}
/** Sends arbitrary bytes (not necessarily a valid encoded frame) on a fresh stream of `kind`. */
sendRaw(connectionId: bigint, bytes: Uint8Array, kind: number): void {
const streamId = this.runtime.streamOpen(connectionId, kind);
this.runtime.streamSendFrame(streamId, bytes, true);
}
/** Shuts down one accepted connection from this side, as a peer disconnect. */
shutdownConnection(connectionId: bigint, code = 0n): void {
this.runtime.connectionShutdown(connectionId, code);
}
async close(): Promise<void> {
this.runtime.listenerClose(this.listenerId);
await this.runtime.release();
@ -483,3 +484,264 @@ test("trust negative (e): a transport that fails isVerifiedFor is disposed and r
await assert.rejects(manager.connect(target), /did not prove|No realtime state path/i);
assert.equal(disposed, true, "the session manager must dispose a transport that fails isVerifiedFor");
});
test("M3: a real connected transport's isVerifiedFor rejects a generation other than the one it connected with", async (t) => {
if (!nativeBridgeAvailable()) return void t.skip("no native bridge available");
const listener = await RawQuicListener.open();
try {
const target: CultMeshRealtimeTarget = { verseId: "aetheria", authorityRuntimeId: "service:aetheria.daemon" };
const candidate: CultMeshRealtimeCandidate = {
endpoint: `cultmesh-state+quic://127.0.0.1:${listener.boundPort}?cert-sha256=${fixturePinHex()}`,
authorityRuntimeId: target.authorityRuntimeId,
priority: 0,
generation: "route-generation-1",
};
const connector = new CultMeshQuicRealtimeConnector();
const transport = await connector.connect(candidate, target);
try {
assert.equal(
transport.isVerifiedFor(target.verseId, target.authorityRuntimeId, CULTMESH_REALTIME_STATE_PROTOCOL_ID, "route-generation-1"),
true,
"the generation it actually connected with must verify",
);
assert.equal(
transport.isVerifiedFor(target.verseId, target.authorityRuntimeId, CULTMESH_REALTIME_STATE_PROTOCOL_ID, "route-generation-2"),
false,
"a mutated/replayed generation must not verify against a real transport",
);
} finally {
transport.dispose();
}
} finally {
await listener.close();
}
});
test("trust negative (e2): a real transport whose generation no longer matches the freshly verified route is disposed", async (t) => {
if (!nativeBridgeAvailable()) return void t.skip("no native bridge available");
const listener = await RawQuicListener.open();
try {
const target: CultMeshRealtimeTarget = { verseId: "aetheria", authorityRuntimeId: "service:aetheria.daemon" };
const root = generateOdinRoot();
const signedRoute = signRoute({
verseId: target.verseId,
authorityRuntimeId: target.authorityRuntimeId,
endpoint: `cultmesh-state+quic://127.0.0.1:${listener.boundPort}?cert-sha256=${fixturePinHex()}`,
protocolIds: [CULTMESH_REALTIME_STATE_PROTOCOL_ID],
privateKey: root.privateKey,
odinKeyId: root.publicKey.keyId,
generation: "route-generation-1",
});
const trust: CultMeshAuthorityTrustPolicy = { mode: "authenticated-remote", odinRoots: [root.publicKey] };
const realConnector = new CultMeshQuicRealtimeConnector();
let connectedTransport: CultMeshRealtimeTransport | undefined;
// A real transport, connected for real against the raw listener, but
// stamped with a generation different from the one the session manager
// just verified — the way a stale cached transport would look after its
// route rotated underneath it.
const replayConnector: CultMeshRealtimeTransportConnector = {
connectorId: "replay",
priority: 0,
canConnect: (c) => realConnector.canConnect(c),
connect: async (c, t, signal) => {
const accepted = listener.acceptOnce();
const transport = await realConnector.connect({ ...c, generation: "route-generation-1-stale" }, t, signal);
await accepted;
connectedTransport = transport;
return transport;
},
};
const manager = new CultMeshQuicRealtimeSessionManager({
lookupSource: new CultMeshStaticRealtimeLookupSource([verseWithRoute(target.verseId, signedRoute)]),
trust,
connectors: [replayConnector],
});
await assert.rejects(manager.connect(target), /did not prove|No realtime state path/i);
assert.ok(connectedTransport, "a real transport must have connected");
// Real cleanup ran: receiveFrame on the disposed transport rejects.
await assert.rejects(connectedTransport!.receiveFrame(), /disposed/i);
} finally {
await listener.close();
}
});
test("the advertised certificate pin is matched case-insensitively (M2)", async (t) => {
if (!nativeBridgeAvailable()) return void t.skip("no native bridge available");
const listener = await RawQuicListener.open();
try {
const target: CultMeshRealtimeTarget = { verseId: "aetheria", authorityRuntimeId: "service:aetheria.daemon" };
const candidate: CultMeshRealtimeCandidate = {
endpoint: `cultmesh-state+quic://127.0.0.1:${listener.boundPort}?cert-sha256=${fixturePinHex().toUpperCase()}`,
authorityRuntimeId: target.authorityRuntimeId,
priority: 0,
generation: "gen-1",
};
const connector = new CultMeshQuicRealtimeConnector();
const transport = await connector.connect(candidate, target);
transport.dispose();
} finally {
await listener.close();
}
});
// ---------------------------------------------------------------------------
// Fix batch: pump/dispatch isolation (1), peer-shutdown cleanup (3), the
// bounded receive backlog and stream-kind pruning (6).
// ---------------------------------------------------------------------------
test("fix 1 / M4: a stream-kind/delivery mismatch faults only that connection; the pump keeps running", async (t) => {
if (!nativeBridgeAvailable()) return void t.skip("no native bridge available");
const listener = await RawQuicListener.open();
try {
const target: CultMeshRealtimeTarget = { verseId: "aetheria", authorityRuntimeId: "service:aetheria.daemon" };
const endpoint = (): string => `cultmesh-state+quic://127.0.0.1:${listener.boundPort}?cert-sha256=${fixturePinHex()}`;
// A reliable stream carrying a frame whose own delivery byte says
// latest-only: `onStreamFrame` throws, which must fault only this
// connection.
const connectorA = new CultMeshQuicRealtimeConnector();
const acceptedA = listener.acceptOnce();
const transportA = await connectorA.connect(
{ endpoint: endpoint(), authorityRuntimeId: target.authorityRuntimeId, priority: 0, generation: "gen-1" },
target,
);
const connectionIdA = await acceptedA;
listener.sendFrame(connectionIdA, testFrame({ delivery: "latest-only" }), CULTMESH_QUIC_STREAM_RELIABLE);
await assert.rejects(transportA.receiveFrame(), /incompatible delivery semantics/i);
// Future receives on the faulted transport reject immediately, with the
// same error, rather than hanging.
await assert.rejects(transportA.receiveFrame(), /incompatible delivery semantics/i);
// A second, independent connection through the same process-wide runtime
// and pump must still work: the fault above did not bring it down.
const connectorB = new CultMeshQuicRealtimeConnector();
const acceptedB = listener.acceptOnce();
const transportB = await connectorB.connect(
{ endpoint: endpoint(), authorityRuntimeId: target.authorityRuntimeId, priority: 0, generation: "gen-1" },
target,
);
try {
const connectionIdB = await acceptedB;
listener.sendFrame(connectionIdB, testFrame(), CULTMESH_QUIC_STREAM_LATEST_ONLY);
const received = await transportB.receiveFrame();
assert.equal(received.bodyId, "body:aetheria:entities");
} finally {
transportB.dispose();
}
} finally {
await listener.close();
}
});
test("fix 3: a peer-initiated shutdown rejects a pending receive and cleans up exactly once", async (t) => {
if (!nativeBridgeAvailable()) return void t.skip("no native bridge available");
const listener = await RawQuicListener.open();
try {
const target: CultMeshRealtimeTarget = { verseId: "aetheria", authorityRuntimeId: "service:aetheria.daemon" };
const candidate: CultMeshRealtimeCandidate = {
endpoint: `cultmesh-state+quic://127.0.0.1:${listener.boundPort}?cert-sha256=${fixturePinHex()}`,
authorityRuntimeId: target.authorityRuntimeId,
priority: 0,
generation: "gen-1",
};
const connector = new CultMeshQuicRealtimeConnector();
const accepted = listener.acceptOnce();
const transport = await connector.connect(candidate, target);
const connectionId = await accepted;
const pendingReceive = transport.receiveFrame();
listener.shutdownConnection(connectionId);
await assert.rejects(pendingReceive, /closed by the remote peer/i);
// Cleanup already ran: a further explicit dispose is a harmless no-op,
// and a fresh receive rejects immediately instead of hanging forever
// (the leak fix 3 exists for).
transport.dispose();
await assert.rejects(transport.receiveFrame(), /closed by the remote peer/i);
} finally {
await listener.close();
}
});
test("fix 6: a slow latest-only reader accumulates at most one pending frame per key, and an equal generation is dropped", async (t) => {
if (!nativeBridgeAvailable()) return void t.skip("no native bridge available");
const listener = await RawQuicListener.open();
try {
const target: CultMeshRealtimeTarget = { verseId: "aetheria", authorityRuntimeId: "service:aetheria.daemon" };
const candidate: CultMeshRealtimeCandidate = {
endpoint: `cultmesh-state+quic://127.0.0.1:${listener.boundPort}?cert-sha256=${fixturePinHex()}`,
authorityRuntimeId: target.authorityRuntimeId,
priority: 0,
generation: "gen-1",
};
const connector = new CultMeshQuicRealtimeConnector();
const accepted = listener.acceptOnce();
const transport = await connector.connect(candidate, target);
try {
const connectionId = await accepted;
// Three increasing-sequence frames for the same (channel, body),
// published while nothing reads: an unbounded queue would deliver all
// three in order; the coalescing inbox keeps only the newest.
listener.sendFrame(connectionId, testFrame({ producerEpoch: 1n, sequence: 1n }), CULTMESH_QUIC_STREAM_LATEST_ONLY);
listener.sendFrame(connectionId, testFrame({ producerEpoch: 1n, sequence: 2n }), CULTMESH_QUIC_STREAM_LATEST_ONLY);
listener.sendFrame(connectionId, testFrame({ producerEpoch: 1n, sequence: 3n }), CULTMESH_QUIC_STREAM_LATEST_ONLY);
// An equal generation, sent last: it must be dropped outright, not
// queued as a fourth pending frame.
listener.sendFrame(connectionId, testFrame({ producerEpoch: 1n, sequence: 3n }), CULTMESH_QUIC_STREAM_LATEST_ONLY);
// Give the pump time to process all four sends before reading.
await new Promise((resolve) => setTimeout(resolve, 200));
const first = await transport.receiveFrame();
assert.equal(first.sequence, 3n, "only the newest pending frame for the key is kept");
const timeout = new AbortController();
const timer = setTimeout(() => timeout.abort(), 300);
await assert.rejects(transport.receiveFrame(timeout.signal), "nothing else should be pending");
clearTimeout(timer);
} finally {
transport.dispose();
}
} finally {
await listener.close();
}
});
test("fix 6: a stream shutdown prunes its entry so streamKinds does not grow without bound", async (t) => {
if (!nativeBridgeAvailable()) return void t.skip("no native bridge available");
const listener = await RawQuicListener.open();
try {
const target: CultMeshRealtimeTarget = { verseId: "aetheria", authorityRuntimeId: "service:aetheria.daemon" };
const candidate: CultMeshRealtimeCandidate = {
endpoint: `cultmesh-state+quic://127.0.0.1:${listener.boundPort}?cert-sha256=${fixturePinHex()}`,
authorityRuntimeId: target.authorityRuntimeId,
priority: 0,
generation: "gen-1",
};
const connector = new CultMeshQuicRealtimeConnector();
const accepted = listener.acceptOnce();
const transport = await connector.connect(candidate, target);
try {
const connectionId = await accepted;
listener.sendFrame(connectionId, testFrame(), CULTMESH_QUIC_STREAM_LATEST_ONLY);
const received = await transport.receiveFrame();
assert.equal(received.sequence, 1n);
// The server-side stream the frame arrived on is a fresh latest-only
// stream the listener opened and finished with `fin: true`; the native
// bridge emits its own shutdown event once MsQuic completes teardown.
// Reaching a second, unrelated frame afterwards is proof the consumer
// kept working, which is what this fix protects: it does not assert
// the internal map directly, since nothing here exposes it.
listener.sendFrame(connectionId, testFrame({ sequence: 2n }), CULTMESH_QUIC_STREAM_LATEST_ONLY);
const second = await transport.receiveFrame();
assert.equal(second.sequence, 2n);
} finally {
transport.dispose();
}
} finally {
await listener.close();
}
});

25
packages/cultmesh-ts/test/realtime-quic-native.test.ts
View file

@ -25,26 +25,13 @@ import {
CULTMESH_QUIC_EVENT_STREAM_SEND_COMPLETE,
CULTMESH_QUIC_STREAM_RELIABLE,
} from "../src/realtime-quic-native";
import { nativeBridgeAvailable } from "./support/native-bridge";
// __dirname at runtime is dist-test/test; fixtures are binary and are never
// compiled/copied there, so they are read from their source location, two
// levels up.
const FIXTURE_P12 = join(__dirname, "..", "..", "test", "fixtures", "quic-test.p12");
function nativeBridgeAvailable(): boolean {
const dir = process.env.CULTMESH_QUIC_NATIVE_DIR;
if (!dir) return false;
try {
const bridge = process.platform === "win32" ? "gamecult_mesh_quic_native.dll" : "libgamecult_mesh_quic_native.so";
const dependency = process.platform === "win32" ? "msquic.dll" : "libmsquic.so.2";
readFileSync(join(dir, bridge));
readFileSync(join(dir, dependency));
return true;
} catch {
return false;
}
}
async function waitFor<T>(collect: (resolve: (value: T) => void) => void, timeoutMs = 5_000): Promise<T> {
return await new Promise<T>((resolve, reject) => {
const timer = setTimeout(() => reject(new Error("timed out waiting for a native QUIC event")), timeoutMs);
@ -139,6 +126,16 @@ test("a full loopback connection delivers one raw frame end to end", async (t) =
}
});
test("fix 2: release() refuses a call with no outstanding reference", async (t) => {
if (!nativeBridgeAvailable()) {
t.skip("CULTMESH_QUIC_NATIVE_DIR is not set to a built native bridge.");
return;
}
const runtime = await CultMeshQuicNativeRuntime.open();
await runtime.release();
await assert.rejects(runtime.release(), /no outstanding reference/i);
});
test("negative grep: no koffi callbacks, .async only on nextEvent", () => {
// __dirname at runtime is dist-test/test; the source lives two levels up,
// in src/, since only .js output is mirrored under dist-test.

261
packages/cultmesh-ts/test/realtime-quic-session-manager.test.ts Normal file
View file

@ -0,0 +1,261 @@
// `CultMeshQuicRealtimeSessionManager`'s own rules, isolated from the native
// bridge with fake connectors/transports: eviction on transport death (fix
// 4), single-flight (fix 4), the disposed guard (fix 4), first-success racing
// with abort of the losers (fix 5), the `protocolIds` filter (M7), the
// `authorityRuntimeId` filter (M8), the session key including
// `authorityRuntimeId` (M9), and the `maxRacedCandidates` cap (M6).
//
// Routes use `trust.mode = "local-development"` over loopback endpoints, so
// `verifyAuthorityRoute` short-circuits without a signed certificate
// (`cultmesh-authority.ts:99`): these tests are about the session manager's
// own binding/racing/caching rules, not route trust, which
// `realtime-quic-consumer.test.ts`'s trust negatives already pin.
import assert from "node:assert/strict";
import test from "node:test";
import type { CultMeshAuthorityTrustPolicy } from "cultnet-ts";
import type { CultMeshAuthorityRouteMessage, CultMeshVerseDescriptorMessage } from "cultnet-ts/contracts";
import {
CultMeshQuicRealtimeSessionManager,
CultMeshStaticRealtimeLookupSource,
CULTMESH_REALTIME_STATE_PROTOCOL_ID,
type CultMeshRealtimeCandidate,
type CultMeshRealtimeTarget,
type CultMeshRealtimeTransport,
type CultMeshRealtimeTransportConnector,
} from "../src/realtime-quic";
const trust: CultMeshAuthorityTrustPolicy = { mode: "local-development", odinRoots: [] };
function route(overrides: Partial<CultMeshAuthorityRouteMessage> = {}): CultMeshAuthorityRouteMessage {
return {
authorityRuntimeId: "service:aetheria.daemon",
endpoint: "cultmesh-state+quic://127.0.0.1:9443",
protocolIds: [CULTMESH_REALTIME_STATE_PROTOCOL_ID],
priority: 0,
generation: "gen-1",
...overrides,
};
}
function verseWithRoutes(verseId: string, routes: readonly CultMeshAuthorityRouteMessage[]): CultMeshVerseDescriptorMessage {
return {
verseId,
displayName: verseId,
authorityModel: "operator-cluster",
compatibility: { transportVersion: "cultmesh.v1", rulesHash: "test", compatibleVerseIds: [], requiredPluginIds: [], optionalPluginIds: [] },
discoveryEndpoints: routes.map((r) => r.endpoint),
authorityRuntimeIds: [...new Set(routes.map((r) => r.authorityRuntimeId))],
authorityRoutes: [...routes],
};
}
interface FakeTransportHandle {
readonly transport: CultMeshRealtimeTransport;
disposed(): boolean;
}
/** A transport that never resolves `receiveFrame` and tracks its own disposal. */
function fakeTransport(endpoint: string, verified = true): FakeTransportHandle {
let disposed = false;
const disposalHandlers: Array<() => void> = [];
const transport: CultMeshRealtimeTransport = {
transportId: "fake",
endpoint,
async sendFrame() {},
async receiveFrame() {
return await new Promise<never>(() => {});
},
dispose() {
if (disposed) return;
disposed = true;
for (const handler of disposalHandlers) handler();
},
isVerifiedFor: () => verified,
onDisposed(handler) {
if (disposed) {
handler();
return;
}
disposalHandlers.push(handler);
},
};
return { transport, disposed: () => disposed };
}
/** A connector that resolves immediately (unless `delayMs` is given) with one fake transport per call. */
function spyConnector(
connectorId: string,
priority: number,
options: { delayMs?: number; fails?: boolean } = {},
): { connector: CultMeshRealtimeTransportConnector; calls: CultMeshRealtimeCandidate[]; handles: FakeTransportHandle[] } {
const calls: CultMeshRealtimeCandidate[] = [];
const handles: FakeTransportHandle[] = [];
const connector: CultMeshRealtimeTransportConnector = {
connectorId,
priority,
canConnect: () => true,
connect: async (candidate, _target, signal) => {
calls.push(candidate);
if (options.delayMs) {
await new Promise<void>((resolve, reject) => {
const timer = setTimeout(resolve, options.delayMs);
signal?.addEventListener("abort", () => {
clearTimeout(timer);
reject(new Error(`${connectorId} connect aborted`));
});
});
}
if (options.fails) throw new Error(`${connectorId} refused to connect`);
const handle = fakeTransport(candidate.endpoint);
handles.push(handle);
return handle.transport;
},
};
return { connector, calls, handles };
}
const target: CultMeshRealtimeTarget = { verseId: "aetheria", authorityRuntimeId: "service:aetheria.daemon" };
test("fix 4: connect() caches a session and evicts it once the transport dies, so the next connect redials", async () => {
const { connector, calls } = spyConnector("fake", 0);
const manager = new CultMeshQuicRealtimeSessionManager({
lookupSource: new CultMeshStaticRealtimeLookupSource([verseWithRoutes(target.verseId, [route()])]),
trust,
connectors: [connector],
});
const first = await manager.connect(target);
assert.equal(await manager.connect(target), first, "a live session is returned from cache, not redialed");
assert.equal(calls.length, 1);
first.dispose();
const second = await manager.connect(target);
assert.notEqual(second, first, "a dead session is evicted, so the next connect redials");
assert.equal(calls.length, 2);
});
test("fix 4: concurrent connect() calls for the same target share one dial (single-flight)", async () => {
const { connector, calls } = spyConnector("fake", 0, { delayMs: 50 });
const manager = new CultMeshQuicRealtimeSessionManager({
lookupSource: new CultMeshStaticRealtimeLookupSource([verseWithRoutes(target.verseId, [route()])]),
trust,
connectors: [connector],
});
const [a, b, c] = await Promise.all([manager.connect(target), manager.connect(target), manager.connect(target)]);
assert.equal(a, b);
assert.equal(b, c);
assert.equal(calls.length, 1, "three concurrent callers must share one dial");
});
test("fix 4: a connect() that completes after dispose() releases its transport instead of caching it", async () => {
const { connector, handles } = spyConnector("fake", 0, { delayMs: 50 });
const manager = new CultMeshQuicRealtimeSessionManager({
lookupSource: new CultMeshStaticRealtimeLookupSource([verseWithRoutes(target.verseId, [route()])]),
trust,
connectors: [connector],
});
const pending = manager.connect(target);
manager.dispose();
await assert.rejects(pending, /disposed/i);
assert.equal(handles.length, 1);
assert.equal(handles[0]!.disposed(), true, "a transport that lands after dispose() must not be kept alive");
await assert.rejects(manager.connect(target), /disposed/i, "a disposed manager refuses further connects");
});
test("fix 5: the race returns the first verified success and aborts the losers", async () => {
const winner = spyConnector("fast", 0, { delayMs: 10 });
const loser = spyConnector("slow", 0, { delayMs: 5_000 });
const manager = new CultMeshQuicRealtimeSessionManager({
lookupSource: new CultMeshStaticRealtimeLookupSource([
verseWithRoutes(target.verseId, [
route({ endpoint: "cultmesh-state+quic://127.0.0.1:9001" }),
route({ endpoint: "cultmesh-state+quic://127.0.0.1:9002" }),
]),
]),
trust,
connectors: [winner.connector, loser.connector],
maxRacedCandidates: 2,
});
const startedAt = Date.now();
const transport = await manager.connect(target);
const elapsedMs = Date.now() - startedAt;
assert.ok(elapsedMs < 1_000, `connect() must return on the first success, not wait out the slow route (took ${elapsedMs}ms)`);
assert.equal(transport.endpoint, "cultmesh-state+quic://127.0.0.1:9001");
});
test("M6: maxRacedCandidates caps how many candidates in a tier are attempted", async () => {
const spy = spyConnector("fake", 0);
const manager = new CultMeshQuicRealtimeSessionManager({
lookupSource: new CultMeshStaticRealtimeLookupSource([
verseWithRoutes(target.verseId, [
route({ endpoint: "cultmesh-state+quic://127.0.0.1:9001", priority: 0 }),
route({ endpoint: "cultmesh-state+quic://127.0.0.1:9002", priority: 1 }),
route({ endpoint: "cultmesh-state+quic://127.0.0.1:9003", priority: 2 }),
]),
]),
trust,
connectors: [spy.connector],
maxRacedCandidates: 1,
});
await manager.connect(target);
assert.equal(spy.calls.length, 1, "only the top maxRacedCandidates entries in the tier are attempted");
assert.equal(spy.calls[0]!.endpoint, "cultmesh-state+quic://127.0.0.1:9001", "the lowest route priority races first");
});
test("M7: a route whose protocolIds omit the realtime-state protocol is not a candidate", async () => {
const spy = spyConnector("fake", 0);
const manager = new CultMeshQuicRealtimeSessionManager({
lookupSource: new CultMeshStaticRealtimeLookupSource([
verseWithRoutes(target.verseId, [route({ protocolIds: ["cultmesh.other_protocol.v1"] })]),
]),
trust,
connectors: [spy.connector],
});
await assert.rejects(manager.connect(target), /No realtime state route was advertised/);
assert.equal(spy.calls.length, 0);
});
test("M8: a route advertised for a different authorityRuntimeId is not a candidate", async () => {
const spy = spyConnector("fake", 0);
const manager = new CultMeshQuicRealtimeSessionManager({
lookupSource: new CultMeshStaticRealtimeLookupSource([
verseWithRoutes(target.verseId, [route({ authorityRuntimeId: "service:other.daemon" })]),
]),
trust,
connectors: [spy.connector],
});
await assert.rejects(manager.connect(target), /No realtime state route was advertised/);
assert.equal(spy.calls.length, 0);
});
test("M9: the session key includes authorityRuntimeId, so two authorities in the same Verse get separate sessions", async () => {
const spy = spyConnector("fake", 0);
const targetA: CultMeshRealtimeTarget = { verseId: "aetheria", authorityRuntimeId: "service:a.daemon" };
const targetB: CultMeshRealtimeTarget = { verseId: "aetheria", authorityRuntimeId: "service:b.daemon" };
const manager = new CultMeshQuicRealtimeSessionManager({
lookupSource: new CultMeshStaticRealtimeLookupSource([
verseWithRoutes("aetheria", [
route({ authorityRuntimeId: "service:a.daemon", endpoint: "cultmesh-state+quic://127.0.0.1:9001" }),
route({ authorityRuntimeId: "service:b.daemon", endpoint: "cultmesh-state+quic://127.0.0.1:9002" }),
]),
]),
trust,
connectors: [spy.connector],
});
const sessionA = await manager.connect(targetA);
const sessionB = await manager.connect(targetB);
assert.notEqual(sessionA, sessionB);
assert.equal(sessionA.endpoint, "cultmesh-state+quic://127.0.0.1:9001");
assert.equal(sessionB.endpoint, "cultmesh-state+quic://127.0.0.1:9002");
assert.equal(spy.calls.length, 2);
});

20
packages/cultmesh-ts/test/support/native-bridge.ts Normal file
View file

@ -0,0 +1,20 @@
// Shared by every QUIC test file that skips instead of failing when no native
// bridge is built: true only when both the bridge and its `msquic` dependency
// are present under `CULTMESH_QUIC_NATIVE_DIR`.
import { readFileSync } from "node:fs";
import { join } from "node:path";
export function nativeBridgeAvailable(): boolean {
const dir = process.env.CULTMESH_QUIC_NATIVE_DIR;
if (!dir) return false;
try {
const bridge = process.platform === "win32" ? "gamecult_mesh_quic_native.dll" : "libgamecult_mesh_quic_native.so";
const dependency = process.platform === "win32" ? "msquic.dll" : "libmsquic.so.2";
readFileSync(join(dir, bridge));
readFileSync(join(dir, dependency));
return true;
} catch {
return false;
}
}