From 047c55c45320151a2c337cbe5bd1ff00464aadaa Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 23 Sep 2026 14:38:46 +0200 Subject: [PATCH] Add tests pinning fixes 1-6, M2, M3, M6-M9, and dedupe the two nativeBridgeAvailable helpers 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 --- .../test/realtime-quic-consumer.test.ts | 286 +++++++++++++++++- .../test/realtime-quic-native.test.ts | 25 +- .../realtime-quic-session-manager.test.ts | 261 ++++++++++++++++ .../cultmesh-ts/test/support/native-bridge.ts | 20 ++ 4 files changed, 566 insertions(+), 26 deletions(-) create mode 100644 packages/cultmesh-ts/test/realtime-quic-session-manager.test.ts create mode 100644 packages/cultmesh-ts/test/support/native-bridge.ts diff --git a/packages/cultmesh-ts/test/realtime-quic-consumer.test.ts b/packages/cultmesh-ts/test/realtime-quic-consumer.test.ts index 8118d5a8..b6048b0c 100644 --- a/packages/cultmesh-ts/test/realtime-quic-consumer.test.ts +++ b/packages/cultmesh-ts/test/realtime-quic-consumer.test.ts @@ -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 { 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(); + } +}); diff --git a/packages/cultmesh-ts/test/realtime-quic-native.test.ts b/packages/cultmesh-ts/test/realtime-quic-native.test.ts index 8646fbe1..f5ffb7b4 100644 --- a/packages/cultmesh-ts/test/realtime-quic-native.test.ts +++ b/packages/cultmesh-ts/test/realtime-quic-native.test.ts @@ -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(collect: (resolve: (value: T) => void) => void, timeoutMs = 5_000): Promise { return await new Promise((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. diff --git a/packages/cultmesh-ts/test/realtime-quic-session-manager.test.ts b/packages/cultmesh-ts/test/realtime-quic-session-manager.test.ts new file mode 100644 index 00000000..9fb916a2 --- /dev/null +++ b/packages/cultmesh-ts/test/realtime-quic-session-manager.test.ts @@ -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 { + 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(() => {}); + }, + 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((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); +}); diff --git a/packages/cultmesh-ts/test/support/native-bridge.ts b/packages/cultmesh-ts/test/support/native-bridge.ts new file mode 100644 index 00000000..74929836 --- /dev/null +++ b/packages/cultmesh-ts/test/support/native-bridge.ts @@ -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; + } +}