From 4f69d116444bb0ba411065d7f980c632014ceac5 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 16 Jun 2026 14:22:41 +0100 Subject: [PATCH] Add C# schema client transport port --- docs/cultnet-transport-parity.md | 13 +- docs/runtime-parity-scope.md | 8 +- src/GameCult.Mesh/CultMeshPeerExchange.cs | 19 +- src/GameCult.Mesh/CultMeshVerseDiscovery.cs | 19 +- .../CultNetSchemaClient.cs | 99 +++++++++ .../CultNetShardReplication.cs | 38 ++-- .../CultNetShardWriteForwarding.cs | 19 +- .../NetworkingTests.cs | 203 ++++++++++++++++++ 8 files changed, 382 insertions(+), 36 deletions(-) create mode 100644 src/GameCult.Networking/CultNetSchemaClient.cs diff --git a/docs/cultnet-transport-parity.md b/docs/cultnet-transport-parity.md index 13a4d90a..49c7e780 100644 --- a/docs/cultnet-transport-parity.md +++ b/docs/cultnet-transport-parity.md @@ -271,6 +271,12 @@ Current progress: counter and duplicate backoff constants. It still owns LiteNetLib socket mechanics and UI-facing reconnect state; the controller owns retry attempt scheduling. +- C# now has an `ICultNetSchemaClient` service-client port for schema-v0 + request/response bodies. The default adapter is still the LiteNetLib `Client`, + but Verse discovery, peer exchange, shard-log fetch, shard-snapshot fetch, + and shard write forwarding clients consume the port instead of constructing + LiteNetLib directly. This moves transport choice below service payload + ownership without claiming every daemon uses native RUDP yet. - C# now also has a poll-driven `CultNetRudpReconnectLoop` for the native RUDP socket transport. Caller-owned game/service loops report closure, ask `ReconnectIfDue(nowMs)` inside their own scheduler, and the shared controller @@ -342,9 +348,10 @@ Current progress: behind the shared transport ports, then wire any remaining daemon/socket reconnect loops through the shared controller. C# built-in LiteNetLib service wrappers now route through `CultNetServerPeer`, TypeScript exposes a - transport-first TCP peer helper, and the core schema-message RUDP matrix is - live in the TypeScript interop gate; broader service adoption remains the - unclaimed layer. + transport-first TCP peer helper, C# schema-v0 service clients consume + `ICultNetSchemaClient`, and the core schema-message RUDP matrix is live in + the TypeScript interop gate; broader service adoption remains the unclaimed + layer. ## RUDP Packet Contract V0 diff --git a/docs/runtime-parity-scope.md b/docs/runtime-parity-scope.md index 3d60da4d..f0f9416b 100644 --- a/docs/runtime-parity-scope.md +++ b/docs/runtime-parity-scope.md @@ -30,7 +30,7 @@ CultMesh. | Runtime | Role | Claimed parity | Not claimed | | --- | --- | --- | --- | -| C# | Reference implementation and production game/server substrate | CultCache typed documents and SoA, CultNet schema-v0, LiteNetLib production path with adapter profile, shared transport profiles, shared RUDP socket transport, RUDP reconnect policy/controller and reconnect-loop helper, LiteNetLib client reconnect-controller adoption, CultMesh node/session/catalog/stream/simulation surfaces | No need to abandon LiteNetLib before the CultNet adapter fully owns production service adoption | +| C# | Reference implementation and production game/server substrate | CultCache typed documents and SoA, CultNet schema-v0, LiteNetLib production path with adapter profile, shared transport profiles, shared RUDP socket transport, RUDP reconnect policy/controller and reconnect-loop helper, LiteNetLib client reconnect-controller adoption, schema-client transport port for service request/response clients, CultMesh node/session/catalog/stream/simulation surfaces | No need to abandon LiteNetLib before the CultNet adapter fully owns production service adoption | | Python | Package runtime, local daemon endpoint, raw interop peer, typed state participant | CultCache/CultNet/CultMesh package surface, schema catalogs, shard catalogs/logs, local TCP CultMesh server, session/prediction/simulation fact helpers, authority leases, streams, RUDP codec/session/socket transport, RUDP reconnect policy/controller and reconnect-loop helper, schema-v0 MessagePack over RUDP evidence, benchmark evidence against C# cache paths | Universal C# throughput, full C# daemon/server stack, and RUDP as the only public service pipe | | TypeScript | Node/browser-adjacent projection and interop runtime | CultCache parity tests, CultNet schema-v0 peer, schema/shard catalog request/sync helpers, transport profiles, TCP-framed transport port and peer helper, shared RUDP socket transport, RUDP reconnect policy/controller, RUDP reconnect-loop helper, local CultMesh node/catalog/authority/stream ergonomics | Full C# game session/server stack | | Rust | Low-level cache/transport/runtime substrate | CultCache/CultNet interop lane, transport profiles, TCP-framed port, shared RUDP codec/session/socket transport, RUDP reconnect policy/controller and reconnect-loop helper, schema-v0 MessagePack over RUDP evidence, small CultMesh facade for peer cards, authority leases, RUDP endpoint parsing, and peer-card RUDP client construction | Full CultMesh game-session/server facade, stream catalog, or production game-session server | @@ -60,8 +60,10 @@ CultMesh. - Broaden production service adoption so remaining daemon paths use the shared transport port consistently. The C# built-in LiteNetLib service wrappers now route through transport-aware peer context, and TypeScript TCP peer creation - has a transport-first helper; broader daemon adoption remains the unclaimed - layer. + has a transport-first helper. C# Verse discovery, peer exchange, shard-log + fetch, shard-snapshot fetch, and shard write forwarding clients now consume a + schema-client transport port with the LiteNetLib client as the default + adapter; broader daemon adoption remains the unclaimed layer. - Extend RUDP adoption through daemon/socket reconnect loops. RUDP profiles now advertise the portable reconnect policy, runtime-local tests prove the shared controller across the targeted runtimes, and the C# LiteNetLib client plus diff --git a/src/GameCult.Mesh/CultMeshPeerExchange.cs b/src/GameCult.Mesh/CultMeshPeerExchange.cs index 8c8a81ed..f53808d2 100644 --- a/src/GameCult.Mesh/CultMeshPeerExchange.cs +++ b/src/GameCult.Mesh/CultMeshPeerExchange.cs @@ -315,6 +315,11 @@ namespace GameCult.Mesh /// Gets or sets a callback used to customize each ephemeral exchange client. /// public Action? ConfigureClient { get; set; } + + /// + /// Gets or sets the schema-v0 client factory. Defaults to the C# LiteNetLib adapter. + /// + public Func? CreateClient { get; set; } } /// @@ -348,11 +353,7 @@ namespace GameCult.Mesh var completion = new TaskCompletionSource( TaskCreationOptions.RunContinuationsAsynchronously); - using var client = new Client(_options.Security ?? ClientSecurityOptions.Development()) - { - AllowUnverifiedCultNetMessages = true - }; - _options.ConfigureClient?.Invoke(client); + using var client = CreateClient(); client.OnCultNet(response => { if (string.Equals(response.MessageId, messageId, StringComparison.Ordinal)) @@ -408,7 +409,13 @@ namespace GameCult.Mesh return count; } - private async Task WaitForConnectionAsync(Client client, string endpoint) + private ICultNetSchemaClient CreateClient() + { + return _options.CreateClient?.Invoke() + ?? CultNetSchemaClients.CreateLiteNetLib(_options.Security, _options.ConfigureClient); + } + + private async Task WaitForConnectionAsync(ICultNetSchemaClient client, string endpoint) { var deadline = DateTimeOffset.UtcNow + _options.ConnectTimeout; while (!client.Connected) diff --git a/src/GameCult.Mesh/CultMeshVerseDiscovery.cs b/src/GameCult.Mesh/CultMeshVerseDiscovery.cs index 2cd7e5d0..d89d3f3d 100644 --- a/src/GameCult.Mesh/CultMeshVerseDiscovery.cs +++ b/src/GameCult.Mesh/CultMeshVerseDiscovery.cs @@ -161,6 +161,11 @@ namespace GameCult.Mesh /// Gets or sets a callback used to customize each ephemeral discovery client. /// public Action? ConfigureClient { get; set; } + + /// + /// Gets or sets the schema-v0 client factory. Defaults to the C# LiteNetLib adapter. + /// + public Func? CreateClient { get; set; } } /// @@ -191,11 +196,7 @@ namespace GameCult.Mesh var completion = new TaskCompletionSource( TaskCreationOptions.RunContinuationsAsynchronously); - using var client = new Client(_options.Security ?? ClientSecurityOptions.Development()) - { - AllowUnverifiedCultNetMessages = true - }; - _options.ConfigureClient?.Invoke(client); + using var client = CreateClient(); client.OnCultNet(response => { if (string.Equals(response.MessageId, messageId, StringComparison.Ordinal)) @@ -243,7 +244,13 @@ namespace GameCult.Mesh return count; } - private async Task WaitForConnectionAsync(Client client, string endpoint) + private ICultNetSchemaClient CreateClient() + { + return _options.CreateClient?.Invoke() + ?? CultNetSchemaClients.CreateLiteNetLib(_options.Security, _options.ConfigureClient); + } + + private async Task WaitForConnectionAsync(ICultNetSchemaClient client, string endpoint) { var deadline = DateTimeOffset.UtcNow + _options.ConnectTimeout; while (!client.Connected) diff --git a/src/GameCult.Networking/CultNetSchemaClient.cs b/src/GameCult.Networking/CultNetSchemaClient.cs new file mode 100644 index 00000000..eb113a2a --- /dev/null +++ b/src/GameCult.Networking/CultNetSchemaClient.cs @@ -0,0 +1,99 @@ +using System; + +namespace GameCult.Networking +{ + /// + /// Minimal client port for request/response services that speak CultNet schema-v0 messages. + /// + public interface ICultNetSchemaClient : IDisposable + { + /// + /// Gets whether the underlying transport reports an established connection. + /// + bool Connected { get; } + + /// + /// Connects the transport to a remote endpoint. + /// + void Connect(string host, int port); + + /// + /// Sends one schema-v0 message through the transport. + /// + void SendCultNet(T message) + where T : ICultNetSchemaMessage; + + /// + /// Adds a callback for one schema-v0 response type. + /// + void OnCultNet(Action callback) + where T : ICultNetSchemaMessage; + } + + /// + /// LiteNetLib-backed implementation of the CultNet schema client port. + /// + public sealed class LiteNetLibCultNetSchemaClient : ICultNetSchemaClient + { + private readonly Client _client; + + /// + /// Creates a LiteNetLib schema client adapter. + /// + public LiteNetLibCultNetSchemaClient( + ClientSecurityOptions? security = null, + Action? configureClient = null) + { + _client = new Client(security ?? ClientSecurityOptions.Development()) + { + AllowUnverifiedCultNetMessages = true + }; + configureClient?.Invoke(_client); + } + + /// + public bool Connected => _client.Connected; + + /// + public void Connect(string host, int port) + { + _client.Connect(host, port); + } + + /// + public void SendCultNet(T message) + where T : ICultNetSchemaMessage + { + _client.SendCultNet(message); + } + + /// + public void OnCultNet(Action callback) + where T : ICultNetSchemaMessage + { + _client.OnCultNet(callback); + } + + /// + public void Dispose() + { + _client.Dispose(); + } + } + + /// + /// Factory helpers for schema client adapters. + /// + public static class CultNetSchemaClients + { + /// + /// Creates the default C# LiteNetLib schema client adapter. + /// + public static ICultNetSchemaClient CreateLiteNetLib( + ClientSecurityOptions? security = null, + Action? configureClient = null) + { + return new LiteNetLibCultNetSchemaClient(security, configureClient); + } + } +} diff --git a/src/GameCult.Networking/CultNetShardReplication.cs b/src/GameCult.Networking/CultNetShardReplication.cs index 2524e0c2..1d81ebbe 100644 --- a/src/GameCult.Networking/CultNetShardReplication.cs +++ b/src/GameCult.Networking/CultNetShardReplication.cs @@ -393,6 +393,11 @@ namespace GameCult.Networking /// Gets or sets a callback used to customize each ephemeral fetch client. /// public Action? ConfigureClient { get; set; } + + /// + /// Gets or sets the schema-v0 client factory. Defaults to the C# LiteNetLib adapter. + /// + public Func? CreateClient { get; set; } } /// @@ -423,11 +428,7 @@ namespace GameCult.Networking var completion = new TaskCompletionSource( TaskCreationOptions.RunContinuationsAsynchronously); - using var client = new Client(_options.Security ?? ClientSecurityOptions.Development()) - { - AllowUnverifiedCultNetMessages = true - }; - _options.ConfigureClient?.Invoke(client); + using var client = CreateClient(); client.OnCultNet(response => { if (string.Equals(response.MessageId, messageId, StringComparison.Ordinal)) @@ -452,7 +453,13 @@ namespace GameCult.Networking return await WaitForResponseAsync(completion.Task, endpoint).ConfigureAwait(false); } - private async Task WaitForConnectionAsync(Client client, string endpoint) + private ICultNetSchemaClient CreateClient() + { + return _options.CreateClient?.Invoke() + ?? CultNetSchemaClients.CreateLiteNetLib(_options.Security, _options.ConfigureClient); + } + + private async Task WaitForConnectionAsync(ICultNetSchemaClient client, string endpoint) { var deadline = DateTimeOffset.UtcNow + _options.ConnectTimeout; while (!client.Connected) @@ -512,6 +519,11 @@ namespace GameCult.Networking /// Gets or sets a callback used to customize each ephemeral fetch client. /// public Action? ConfigureClient { get; set; } + + /// + /// Gets or sets the schema-v0 client factory. Defaults to the C# LiteNetLib adapter. + /// + public Func? CreateClient { get; set; } } /// @@ -539,11 +551,7 @@ namespace GameCult.Networking var completion = new TaskCompletionSource( TaskCreationOptions.RunContinuationsAsynchronously); - using var client = new Client(_options.Security ?? ClientSecurityOptions.Development()) - { - AllowUnverifiedCultNetMessages = true - }; - _options.ConfigureClient?.Invoke(client); + using var client = CreateClient(); client.OnCultNet(response => { if (string.Equals(response.MessageId, messageId, StringComparison.Ordinal)) @@ -567,7 +575,13 @@ namespace GameCult.Networking return await WaitForResponseAsync(completion.Task, endpoint).ConfigureAwait(false); } - private async Task WaitForConnectionAsync(Client client, string endpoint) + private ICultNetSchemaClient CreateClient() + { + return _options.CreateClient?.Invoke() + ?? CultNetSchemaClients.CreateLiteNetLib(_options.Security, _options.ConfigureClient); + } + + private async Task WaitForConnectionAsync(ICultNetSchemaClient client, string endpoint) { var deadline = DateTimeOffset.UtcNow + _options.ConnectTimeout; while (!client.Connected) diff --git a/src/GameCult.Networking/CultNetShardWriteForwarding.cs b/src/GameCult.Networking/CultNetShardWriteForwarding.cs index d5114d69..686621be 100644 --- a/src/GameCult.Networking/CultNetShardWriteForwarding.cs +++ b/src/GameCult.Networking/CultNetShardWriteForwarding.cs @@ -55,6 +55,11 @@ namespace GameCult.Networking /// Gets or sets a callback used to customize each ephemeral forwarding client. /// public Action? ConfigureClient { get; set; } + + /// + /// Gets or sets the schema-v0 client factory. Defaults to the C# LiteNetLib adapter. + /// + public Func? CreateClient { get; set; } } /// @@ -115,17 +120,19 @@ namespace GameCult.Networking private async Task SendAsync(string endpoint, T message) where T : ICultNetSchemaMessage { var (host, port) = ParseEndpoint(endpoint); - using var client = new Client(_options.Security ?? ClientSecurityOptions.Development()) - { - AllowUnverifiedCultNetMessages = true - }; - _options.ConfigureClient?.Invoke(client); + using var client = CreateClient(); client.Connect(host, port); await WaitForConnectionAsync(client, endpoint).ConfigureAwait(false); client.SendCultNet(message); } - private async Task WaitForConnectionAsync(Client client, string endpoint) + private ICultNetSchemaClient CreateClient() + { + return _options.CreateClient?.Invoke() + ?? CultNetSchemaClients.CreateLiteNetLib(_options.Security, _options.ConfigureClient); + } + + private async Task WaitForConnectionAsync(ICultNetSchemaClient client, string endpoint) { var deadline = DateTimeOffset.UtcNow + _options.ConnectTimeout; while (!client.Connected) diff --git a/tests/GameCult.Networking.Tests/NetworkingTests.cs b/tests/GameCult.Networking.Tests/NetworkingTests.cs index d58fc945..e360b5e4 100644 --- a/tests/GameCult.Networking.Tests/NetworkingTests.cs +++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs @@ -1732,6 +1732,151 @@ namespace GameCult.Networking.Tests Assert.That(client, Is.Not.Null); } + [Test] + public async Task CultMeshVerseDiscoveryClient_Uses_SchemaClientPort() + { + var fake = new CapturingSchemaClient(message => + { + var request = (CultMeshVerseCatalogRequestMessage)message; + return new CultMeshVerseCatalogResponseMessage + { + MessageId = request.MessageId, + Verses = + [ + new CultMeshVerseDescriptor( + "local", + "Local Verse", + CultMeshVerseAuthorityModel.OperatorCluster, + new CultMeshVerseCompatibility("cultmesh.v0", "rules"), + discoveryEndpoints: ["cultnet://mesh.example.test:4075"]).ToMessage() + ] + }; + }); + var client = new CultMeshVerseDiscoveryClient(new CultMeshVerseDiscoveryClientOptions + { + CreateClient = () => fake + }); + + var response = await client.FetchAsync("cultnet://mesh.example.test:4075"); + + Assert.That(fake.ConnectedHost, Is.EqualTo("mesh.example.test")); + Assert.That(fake.ConnectedPort, Is.EqualTo(4075)); + Assert.That(fake.SentMessages.Single(), Is.TypeOf()); + Assert.That(response.Verses.Single().VerseId, Is.EqualTo("local")); + } + + [Test] + public async Task CultMeshPeerExchangeClient_Uses_SchemaClientPort() + { + var fake = new CapturingSchemaClient(message => + { + var request = (CultMeshPeerExchangeRequestMessage)message; + return new CultMeshPeerExchangeResponseMessage + { + MessageId = request.MessageId, + Peers = + [ + new CultMeshPeerCard( + "peer-a", + request.VerseId, + ["cultnet://peer-a.example.test:3075"], + roles: ["shard-primary"]).ToMessage() + ] + }; + }); + var client = new CultMeshPeerExchangeClient(new CultMeshPeerExchangeClientOptions + { + CreateClient = () => fake + }); + + var response = await client.FetchAsync( + "cultnet://mesh.example.test:4075", + new CultMeshPeerExchangeRequestMessage { VerseId = "local" }); + + Assert.That(fake.ConnectedHost, Is.EqualTo("mesh.example.test")); + Assert.That(fake.SentMessages.Single(), Is.TypeOf()); + Assert.That(response.Peers.Single().PeerId, Is.EqualTo("peer-a")); + } + + [Test] + public async Task CultNetShardFetchers_Use_SchemaClientPort() + { + var logClient = new CapturingSchemaClient(message => + { + var request = (CultNetShardLogRequestMessage)message; + return new CultNetShardLogResponseMessage + { + MessageId = request.MessageId, + ShardId = request.ShardId, + ShardEpoch = request.ShardEpoch ?? 0, + Entries = [] + }; + }); + var snapshotClient = new CapturingSchemaClient(message => + { + var request = (CultNetSnapshotRequestMessage)message; + return new CultNetSnapshotResponseRawMessage + { + MessageId = request.MessageId, + ShardId = request.ShardId, + ShardEpoch = request.ShardEpoch, + Documents = [] + }; + }); + var shard = new CultNetShardDescriptor( + "notes", + "primary", + 7, + isPrimary: false, + schemaIds: ["note.v0"], + primaryEndpoints: ["cultnet://primary.example.test:4075"]); + + var logFetcher = new CultNetSchemaShardLogFetcher(new CultNetSchemaShardLogFetcherOptions + { + CreateClient = () => logClient + }); + var snapshotFetcher = new CultNetSchemaShardSnapshotFetcher(new CultNetSchemaShardSnapshotFetcherOptions + { + CreateClient = () => snapshotClient + }); + + var log = await logFetcher.FetchAsync(shard, afterSequence: 4, limit: 2); + var snapshot = await snapshotFetcher.FetchAsync(shard); + + Assert.That(logClient.ConnectedHost, Is.EqualTo("primary.example.test")); + Assert.That(logClient.SentMessages.Single(), Is.TypeOf()); + Assert.That(log.ShardId, Is.EqualTo("notes")); + Assert.That(snapshotClient.SentMessages.Single(), Is.TypeOf()); + Assert.That(snapshot.ShardEpoch, Is.EqualTo(7)); + } + + [Test] + public async Task CultNetSchemaWriteForwarder_Uses_SchemaClientPort() + { + var fake = new CapturingSchemaClient(_ => null); + var forwarder = new CultNetSchemaWriteForwarder(new CultNetSchemaWriteForwarderOptions + { + CreateClient = () => fake + }); + var shard = new CultNetShardDescriptor( + "notes", + "primary", + 3, + isPrimary: false, + primaryEndpoints: ["cultnet://primary.example.test:4075"]); + + await forwarder.ForwardDeleteAsync(shard, new CultNetDocumentDeleteMessage + { + SchemaId = "note.v0", + RecordKey = "note:1" + }); + + var sent = (CultNetDocumentDeleteMessage)fake.SentMessages.Single(); + Assert.That(fake.ConnectedHost, Is.EqualTo("primary.example.test")); + Assert.That(sent.ShardId, Is.EqualTo("notes")); + Assert.That(sent.ShardEpoch, Is.EqualTo(3)); + } + [Test] public void CultMeshVerseCatalog_Finds_CompatibleTransferTargets() { @@ -3509,6 +3654,64 @@ namespace GameCult.Networking.Tests } } + private sealed class CapturingSchemaClient : ICultNetSchemaClient + { + private readonly Func _respond; + private readonly Dictionary> _handlers = new Dictionary>(); + + public CapturingSchemaClient(Func respond) + { + _respond = respond; + } + + public bool Connected { get; private set; } + public string? ConnectedHost { get; private set; } + public int ConnectedPort { get; private set; } + public List SentMessages { get; } = new List(); + + public void Connect(string host, int port) + { + Connected = true; + ConnectedHost = host; + ConnectedPort = port; + } + + public void SendCultNet(T message) + where T : ICultNetSchemaMessage + { + SentMessages.Add(message); + var response = _respond(message); + if (response == null) + { + return; + } + + if (_handlers.TryGetValue(response.GetType(), out var handlers)) + { + foreach (var handler in handlers) + { + handler.DynamicInvoke(response); + } + } + } + + public void OnCultNet(Action callback) + where T : ICultNetSchemaMessage + { + if (!_handlers.TryGetValue(typeof(T), out var handlers)) + { + handlers = new List(); + _handlers[typeof(T)] = handlers; + } + + handlers.Add(callback); + } + + public void Dispose() + { + } + } + [MessagePack.MessagePackObject] public sealed class PlayerDataPayload {