diff --git a/packages/cultmesh-ts/test/interop/cultmesh-quic-peer.ts b/packages/cultmesh-ts/test/interop/cultmesh-quic-peer.ts new file mode 100644 index 00000000..909cdd86 --- /dev/null +++ b/packages/cultmesh-ts/test/interop/cultmesh-quic-peer.ts @@ -0,0 +1,107 @@ +// The TypeScript half of the QUIC realtime interop lane: a one-shot process +// that dials an advertised `cultmesh-state+quic://` endpoint with the native +// connector, receives `--expect` frames from a peer (normally the C# managed +// provider started by `cultnet-interop.test.ts`'s +// "CultMesh QUIC realtime" test), and prints them as one JSON line. It never +// imports `cultmesh-ts` itself (the harness drives processes, and the +// dependency direction is `cultmesh-ts` -> `cultnet-ts`, never the reverse); +// this script reaches the connector through its own package's `src/`. + +import process from "node:process"; + +import { CultMeshQuicRealtimeConnector, type CultMeshRealtimeTarget } from "../../src/realtime-quic"; + +function parseArgs(argv: readonly string[]): Map { + const args = new Map(); + for (let index = 0; index < argv.length; index += 1) { + const token = argv[index]; + if (!token.startsWith("--")) continue; + const name = token.slice(2); + const value = argv[index + 1]; + if (!value || value.startsWith("--")) throw new Error(`Missing value for --${name}`); + args.set(name, value); + index += 1; + } + return args; +} + +function requireArg(args: Map, name: string): string { + const value = args.get(name); + if (!value) throw new Error(`Missing required argument --${name}`); + return value; +} + +function optionalIntArg(args: Map, name: string, fallback: number): number { + const raw = args.get(name); + if (!raw) return fallback; + const parsed = Number.parseInt(raw, 10); + if (!Number.isInteger(parsed)) throw new Error(`Argument --${name} must be an integer.`); + return parsed; +} + +function writeJsonLine(value: unknown): void { + process.stdout.write(`${JSON.stringify(value)}\n`); +} + +function writeLog(event: string, payload: Record): void { + process.stderr.write(`${JSON.stringify({ event, ...payload })}\n`); +} + +async function main(): Promise { + const [mode, ...rest] = process.argv.slice(2); + if (mode !== "dial") throw new Error("Expected mode: dial"); + const args = parseArgs(rest); + const endpoint = requireArg(args, "endpoint"); + const expect = optionalIntArg(args, "expect", 1); + const timeoutMs = optionalIntArg(args, "timeout-ms", 15_000); + + const target: CultMeshRealtimeTarget = { verseId: "interop", authorityRuntimeId: "interop.csharp-provider" }; + const connector = new CultMeshQuicRealtimeConnector(); + const transport = await connector.connect( + { endpoint, authorityRuntimeId: target.authorityRuntimeId, priority: 0, generation: "interop" }, + target, + ); + writeLog("connected", { endpoint, transportId: transport.transportId }); + + const frames: Array<{ + channelId: string; + schemaId: string; + bodyId: string; + producerEpoch: string; + sequence: string; + delivery: string; + payloadHex: string; + }> = []; + try { + const deadline = Date.now() + timeoutMs; + while (frames.length < expect) { + const remaining = deadline - Date.now(); + if (remaining <= 0) throw new Error(`Timed out after receiving ${frames.length} of ${expect} frames.`); + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(), remaining); + try { + const frame = await transport.receiveFrame(controller.signal); + frames.push({ + channelId: frame.channelId, + schemaId: frame.schemaId, + bodyId: frame.bodyId, + producerEpoch: frame.producerEpoch.toString(), + sequence: frame.sequence.toString(), + delivery: frame.delivery, + payloadHex: Buffer.from(frame.payload).toString("hex"), + }); + } finally { + clearTimeout(timer); + } + } + } finally { + transport.dispose(); + } + + writeJsonLine({ endpoint, transportId: transport.transportId, frames }); +} + +main().catch((error) => { + writeLog("fatal", { error: error instanceof Error ? (error.stack ?? error.message) : String(error) }); + process.exitCode = 1; +}); diff --git a/packages/cultnet-ts/test/interop/cultnet-interop.test.ts b/packages/cultnet-ts/test/interop/cultnet-interop.test.ts index 51bea1ff..edc3a144 100644 --- a/packages/cultnet-ts/test/interop/cultnet-interop.test.ts +++ b/packages/cultnet-ts/test/interop/cultnet-interop.test.ts @@ -57,6 +57,8 @@ const csharpDllPath = resolve( "net10.0", "GameCult.Networking.InteropPeer.dll", ); +const cultMeshTsRoot = resolve(cultLibRoot, "packages", "cultmesh-ts"); +const cultMeshQuicPeerScript = resolve(cultMeshTsRoot, "dist-test", "test", "interop", "cultmesh-quic-peer.js"); const rustBinaryPath = resolve( cultnetRsRoot, "target", @@ -72,6 +74,7 @@ const processCleanupTimeoutMs = 5_000; let rustInteropPeerBuild: Promise | undefined; let csharpInteropPeerBuild: Promise | undefined; let kotlinInteropPeerBuild: Promise | undefined; +let cultMeshTsInteropPeerBuild: Promise | undefined; function resolvePythonCommand(): string { const codexPythonCommand = join(homedir(), ".cache", "codex-runtimes", "codex-primary-runtime", "dependencies", "python", "python.exe"); @@ -392,6 +395,56 @@ transport.close() raise SystemExit("timed out waiting for TypeScript RUDP schema-v0 message") `; +test("CultMesh QUIC realtime: C# managed provider and TypeScript consumer", async (t) => { + if (!process.env.CULTMESH_QUIC_NATIVE_DIR) { + t.skip("CULTMESH_QUIC_NATIVE_DIR is not set to a built native bridge."); + return; + } + logInteropPhase("quic-realtime", "build peers"); + await buildCSharpInteropPeer(); + await buildCultMeshTsQuicPeer(); + + const servers: RunningServeProcess[] = []; + t.after(async () => { + for (const server of servers) { + server.child.kill(); + } + await delay(processCleanupTimeoutMs); + }); + + logInteropPhase("quic-realtime", "start csharp QUIC provider"); + servers.push(await spawnServeProcess("csharp-quic", { + command: dotnetCommand, + args: [ + csharpDllPath, + "quic-realtime-serve", + "--frames", "5", + "--delivery", "latest-only", + "--interval-ms", "50", + ], + cwd: cultLibRoot, + })); + const ready = (await servers[servers.length - 1].ready) as { status: string; endpoint: string }; + assert.equal(ready.status, "ready"); + assert.match(ready.endpoint, /^cultmesh-state\+quic:\/\/127\.0\.0\.1:\d+\?cert-sha256=[0-9A-Fa-f]{64}$/); + + logInteropPhase("quic-realtime", "typescript dials the csharp QUIC provider"); + const tsDial = await runJsonCommand("ts-quic-dial", process.execPath, [ + cultMeshQuicPeerScript, + "dial", + "--endpoint", ready.endpoint, + "--expect", "5", + ], cultMeshTsRoot, { CULTMESH_QUIC_NATIVE_DIR: process.env.CULTMESH_QUIC_NATIVE_DIR ?? "" }); + + assert.equal(tsDial.transportId, "msquic-realtime"); + assert.equal(tsDial.frames.length, 5); + for (const [index, frame] of tsDial.frames.entries()) { + assert.equal(frame.delivery, "latest-only"); + assert.equal(frame.bodyId, "interop:quic-realtime:frame"); + assert.equal(Buffer.from(frame.payloadHex, "hex").toString("utf8"), `frame-${index + 1}`); + } +}); + test("CultNet TS/Rust/C#/Python peers discover each other and exchange raw state over the shared schema-v0 lane", async (t) => { logInteropPhase("schema-v0", "build peers"); await buildInteropPeers(); @@ -2434,6 +2487,15 @@ async function buildCSharpInteropPeer(): Promise { await csharpInteropPeerBuild; } +async function buildCultMeshTsQuicPeer(): Promise { + cultMeshTsInteropPeerBuild ??= (async () => { + const tsc = resolve(cultLibRoot, "node_modules", "typescript", "bin", "tsc"); + await execFileAsync(process.execPath, [tsc, "-p", "tsconfig.json"], { cwd: cultMeshTsRoot }); + await execFileAsync(process.execPath, [tsc, "-p", "tsconfig.test.json"], { cwd: cultMeshTsRoot }); + })(); + await cultMeshTsInteropPeerBuild; +} + async function buildKotlinInteropPeer(): Promise { kotlinInteropPeerBuild ??= execFileAsync( "powershell", diff --git a/tests/GameCult.Networking.InteropPeer/GameCult.Networking.InteropPeer.csproj b/tests/GameCult.Networking.InteropPeer/GameCult.Networking.InteropPeer.csproj index 2a782a72..e55facc5 100644 --- a/tests/GameCult.Networking.InteropPeer/GameCult.Networking.InteropPeer.csproj +++ b/tests/GameCult.Networking.InteropPeer/GameCult.Networking.InteropPeer.csproj @@ -4,10 +4,12 @@ net10.0 enable GameCult.Networking.InteropPeer + $(NoWarn);CA1416 + diff --git a/tests/GameCult.Networking.InteropPeer/Program.cs b/tests/GameCult.Networking.InteropPeer/Program.cs index 3e1de70b..a7d96139 100644 --- a/tests/GameCult.Networking.InteropPeer/Program.cs +++ b/tests/GameCult.Networking.InteropPeer/Program.cs @@ -5,11 +5,15 @@ using System.IO; using System.Linq; using System.Net; using System.Net.Sockets; +using System.Security.Cryptography; +using System.Security.Cryptography.X509Certificates; using System.Text; using System.Text.Json; using System.Threading; using System.Threading.Tasks; using GameCult.Caching; +using GameCult.Mesh; +using GameCult.Mesh.Quic; using GameCult.Networking; using MessagePack; @@ -21,7 +25,7 @@ static async Task ProgramMainAsync(string[] args) { if (args.Length == 0) { - throw new InvalidOperationException("Expected mode: serve | probe | dial | rudp-serve-once | rudp-dial-once | rudp-serve-message-once | rudp-dial-message-once"); + throw new InvalidOperationException("Expected mode: serve | probe | dial | rudp-serve-once | rudp-dial-once | rudp-serve-message-once | rudp-dial-message-once | quic-realtime-serve"); } var mode = args[0]; @@ -49,6 +53,9 @@ static async Task ProgramMainAsync(string[] args) case "rudp-dial-message-once": RudpDialMessageOnce(options); return 0; + case "quic-realtime-serve": + await QuicRealtimeServeAsync(ParseQuicRealtimeServeConfig(options)); + return 0; default: throw new InvalidOperationException($"Unknown mode {mode}"); } @@ -1213,6 +1220,89 @@ static async Task WaitForeverAsync(CancellationToken cancellationToken) } } +static async Task QuicRealtimeServeAsync(QuicRealtimeServeConfig config) +{ + using var certificate = CreateQuicInteropCertificate(); + await using var server = await CultMeshQuicRealtimeServer.ListenAsync(new CultMeshQuicRealtimeServerOptions + { + ListenEndPoint = new IPEndPoint(IPAddress.Loopback, config.Port), + ServerCertificate = certificate + }); + var pin = Convert.ToHexString(SHA256.HashData(certificate.RawData)); + var endpoint = $"cultmesh-state+quic://127.0.0.1:{server.LocalEndPoint.Port}?cert-sha256={pin}"; + WriteJsonLine(new { status = "ready", endpoint }); + + var delivery = config.Delivery switch + { + "reliable-ordered" => CultMeshRealtimeDelivery.ReliableOrdered, + "latest-only" => CultMeshRealtimeDelivery.LatestOnly, + _ => throw new InvalidOperationException($"Unsupported --delivery '{config.Delivery}'.") + }; + + var stdinClosed = WatchStdinCloseAsync(); + for (var sequence = 1; sequence <= config.Frames && !stdinClosed.IsCompleted; sequence++) + { + var frame = new CultMeshRealtimeFrame + { + ChannelId = "interop.quic-realtime", + SchemaId = "gamecult.interop.quic_realtime_frame.v0", + BodyId = "interop:quic-realtime:frame", + ProducerEpoch = 1, + Sequence = sequence, + Delivery = delivery, + Payload = Encoding.UTF8.GetBytes($"frame-{sequence}") + }; + await server.BroadcastAsync(frame); + WriteLog("quic-realtime-serve", new { sent = sequence }); + if (config.IntervalMs > 0) await Task.Delay(config.IntervalMs); + } +} + +static Task WatchStdinCloseAsync() +{ + return Task.Run(async () => + { + while (await Console.In.ReadLineAsync() != null) + { + } + }); +} + +static QuicRealtimeServeConfig ParseQuicRealtimeServeConfig(Dictionary options) +{ + return new QuicRealtimeServeConfig( + ParseOptionalIntArg(options, "port", 0), + ParseOptionalIntArg(options, "frames", 5), + options.TryGetValue("delivery", out var delivery) ? delivery : "latest-only", + ParseOptionalIntArg(options, "interval-ms", 50)); +} + +static X509Certificate2 CreateQuicInteropCertificate() +{ + using var key = RSA.Create(2048); + var request = new CertificateRequest( + "CN=localhost", + key, + HashAlgorithmName.SHA256, + RSASignaturePadding.Pkcs1); + request.CertificateExtensions.Add(new X509BasicConstraintsExtension(false, false, 0, false)); + request.CertificateExtensions.Add(new X509KeyUsageExtension(X509KeyUsageFlags.DigitalSignature, false)); + request.CertificateExtensions.Add(new X509EnhancedKeyUsageExtension( + new OidCollection { new("1.3.6.1.5.5.7.3.1") }, + false)); + var names = new SubjectAlternativeNameBuilder(); + names.AddDnsName("localhost"); + names.AddIpAddress(IPAddress.Loopback); + request.CertificateExtensions.Add(names.Build()); + request.CertificateExtensions.Add(new X509SubjectKeyIdentifierExtension(request.PublicKey, false)); + using var generated = request.CreateSelfSigned( + DateTimeOffset.UtcNow.AddMinutes(-1), + DateTimeOffset.UtcNow.AddHours(1)); + return X509CertificateLoader.LoadPkcs12(generated.Export(X509ContentType.Pfx), null); +} + +sealed record QuicRealtimeServeConfig(int Port, int Frames, string Delivery, int IntervalMs); + static Dictionary ParseArgs(string[] args) { var parsed = new Dictionary(StringComparer.OrdinalIgnoreCase);