From e4d2446dbe4dd268ee3b7bd20f686d72e66a53e3 Mon Sep 17 00:00:00 2001 From: Libby Date: Sun, 8 Nov 2026 22:06:55 +0000 Subject: [PATCH] C# InteropPeer: receipt-serve (CultNetDatabaseServer on an ephemeral port) and receipt-relay (LiteNetLib byte shim for the TS client) Co-Authored-By: Claude Sonnet 5.5 --- .../Program.cs | 8 +- .../ReceiptHarness.cs | 118 ++++++++++++++++++ 2 files changed, 125 insertions(+), 1 deletion(-) create mode 100644 tests/GameCult.Networking.InteropPeer/ReceiptHarness.cs diff --git a/tests/GameCult.Networking.InteropPeer/Program.cs b/tests/GameCult.Networking.InteropPeer/Program.cs index 17a1ea89..fcfcf09c 100644 --- a/tests/GameCult.Networking.InteropPeer/Program.cs +++ b/tests/GameCult.Networking.InteropPeer/Program.cs @@ -25,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 | quic-realtime-serve"); + 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 | receipt-serve | receipt-relay"); } var mode = args[0]; @@ -53,6 +53,12 @@ static async Task ProgramMainAsync(string[] args) case "rudp-dial-message-once": RudpDialMessageOnce(options); return 0; + case "receipt-serve": + await GameCult.Networking.InteropPeer.ReceiptHarness.ServeAsync(); + return 0; + case "receipt-relay": + await GameCult.Networking.InteropPeer.ReceiptHarness.RelayAsync(int.Parse(options["port"], CultureInfo.InvariantCulture)); + return 0; case "quic-realtime-serve": await QuicRealtimeServeAsync(ParseQuicRealtimeServeConfig(options)); return 0; diff --git a/tests/GameCult.Networking.InteropPeer/ReceiptHarness.cs b/tests/GameCult.Networking.InteropPeer/ReceiptHarness.cs new file mode 100644 index 00000000..75eaf4da --- /dev/null +++ b/tests/GameCult.Networking.InteropPeer/ReceiptHarness.cs @@ -0,0 +1,118 @@ +using System; +using System.IO; +using System.Text.Json; +using System.Threading; +using System.Threading.Tasks; +using GameCult.Caching; +using GameCult.Networking; +using LiteNetLib; +using R3; + +// The real C# receipt server (CultNetDatabaseServer on Server) and the one transport shim the TypeScript client needs to reach it, +// for the cross-runtime write-receipt interop harness (packages/cultmesh-ts/test/interop/write-receipt-interop.test.ts). +// TypeScript has no LiteNetLib client, so the shim moves bytes and nothing else: each CultNet frame is a LiteNetLib reliable +// ordered payload, and on the pipe it is a 4-byte big-endian length followed by the frame. +namespace GameCult.Networking.InteropPeer +{ + [CultDocument("receipt.interop_note", ReceiptHarness.SchemaId)] + [MessagePack.MessagePackObject] + public sealed class ReceiptInteropNote + { + [MessagePack.Key(0)] + public string Text { get; set; } = string.Empty; + } + + internal static class ReceiptHarness + { + public const string SchemaId = "receipt.interop_note.v1"; + public const string ShardId = "receipt-interop-shard"; + // A record key with this prefix is refused by the authority: how the harness asks for a refused write. + public const string RefusedPrefix = "refuse:"; + + private sealed class Authority : ICultNetRemoteWriteAuthority + { + public CultNetRemoteWriteDecision Decide(CultNetRemoteWriteRequest request) => + request.RecordKey.ToString().StartsWith(RefusedPrefix, StringComparison.Ordinal) + ? CultNetRemoteWriteDecision.Refuse("denied") + : CultNetRemoteWriteDecision.Grant(); + } + + private static void Emit(object fact) + { + lock (Console.Out) Console.Out.WriteLine(JsonSerializer.Serialize(fact)); + } + + // stdout: {"event":"listening","port":N} then {"event":"applied"|"removed","recordKey":"..."} per committed change; runs until stdin closes. + public static async Task ServeAsync() + { + var cache = new CultCache(); + var database = new CultNetDatabase(cache, new CultNetDatabaseOptions + { + Shards = [new CultNetShardDescriptor(ShardId, "primary", epoch: 1, isPrimary: true, schemaIds: [SchemaId])] + }); + using var server = new Server(cache, ServerSecurityOptions.Development()) { Port = 0 }; + using var databaseServer = new CultNetDatabaseServer(server, database, new CultNetDatabaseServerOptions + { + RemoteWriteAuthority = new Authority() + }); + using var watch = cache.Watch().Subscribe(change => + Emit(new { @event = change.Kind == CultCacheDocumentChangeKind.Removed ? "removed" : "applied", recordKey = change.Key.ToString() })); + server.Start(); + Emit(new { @event = "listening", port = server.Port, shardId = ShardId, schemaId = SchemaId }); + await Task.Run(() => Console.In.ReadToEnd()); + } + + // Connects to the server on `port` and relays frames between stdin/stdout and the connection; exits when either ends. + public static async Task RelayAsync(int port) + { + var listener = new EventBasedNetListener(); + var connected = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var ended = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var output = Console.OpenStandardOutput(); + var gate = new object(); + listener.PeerConnectedEvent += _ => connected.TrySetResult(true); + listener.PeerDisconnectedEvent += (_, _) => ended.TrySetResult(true); + listener.NetworkReceiveEvent += (_, reader, _, _) => + { + var frame = reader.GetRemainingBytes(); + var header = new byte[] { (byte)(frame.Length >> 24), (byte)(frame.Length >> 16), (byte)(frame.Length >> 8), (byte)frame.Length }; + lock (gate) + { + output.Write(header); + output.Write(frame); + output.Flush(); + } + }; + var manager = new NetManager(listener) { UnsyncedEvents = true }; + manager.Start(0); + var peer = manager.Connect("127.0.0.1", port, ServerSecurityOptions.Development().ConnectionKey); + if (await Task.WhenAny(connected.Task, Task.Delay(TimeSpan.FromSeconds(10))) != connected.Task) + throw new InvalidOperationException("the relay did not connect"); + var input = Console.OpenStandardInput(); + var pump = Task.Run(() => + { + var header = new byte[4]; + while (ReadExactly(input, header)) + { + var frame = new byte[(header[0] << 24) | (header[1] << 16) | (header[2] << 8) | header[3]]; + if (!ReadExactly(input, frame)) break; + peer.Send(frame, DeliveryMethod.ReliableOrdered); + } + }); + await Task.WhenAny(pump, ended.Task); + manager.Stop(); + } + + private static bool ReadExactly(Stream stream, byte[] buffer) + { + var read = 0; + while (read < buffer.Length) + { + var count = stream.Read(buffer, read, buffer.Length - read); + if (count <= 0) return false; + read += count; + } + return true; + } + } +}