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

Add C# schema client transport port

Browse source
This commit is contained in:
Meta Vens 2026-06-16 14:22:41 +01:00
commit 4f69d11644
8 changed files with 382 additions and 36 deletions
Download patch file Download diff file Expand all files Collapse all files

13
docs/cultnet-transport-parity.md
View file

@ -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

8
docs/runtime-parity-scope.md
View file

@ -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

19
src/GameCult.Mesh/CultMeshPeerExchange.cs
View file

@ -315,6 +315,11 @@ namespace GameCult.Mesh
/// Gets or sets a callback used to customize each ephemeral exchange client.
/// </summary>
public Action<Client>? ConfigureClient { get; set; }
/// <summary>
/// Gets or sets the schema-v0 client factory. Defaults to the C# LiteNetLib adapter.
/// </summary>
public Func<ICultNetSchemaClient>? CreateClient { get; set; }
}
/// <summary>
@ -348,11 +353,7 @@ namespace GameCult.Mesh
var completion = new TaskCompletionSource<CultMeshPeerExchangeResponseMessage>(
TaskCreationOptions.RunContinuationsAsynchronously);
using var client = new Client(_options.Security ?? ClientSecurityOptions.Development())
{
AllowUnverifiedCultNetMessages = true
};
_options.ConfigureClient?.Invoke(client);
using var client = CreateClient();
client.OnCultNet<CultMeshPeerExchangeResponseMessage>(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)

19
src/GameCult.Mesh/CultMeshVerseDiscovery.cs
View file

@ -161,6 +161,11 @@ namespace GameCult.Mesh
/// Gets or sets a callback used to customize each ephemeral discovery client.
/// </summary>
public Action<Client>? ConfigureClient { get; set; }
/// <summary>
/// Gets or sets the schema-v0 client factory. Defaults to the C# LiteNetLib adapter.
/// </summary>
public Func<ICultNetSchemaClient>? CreateClient { get; set; }
}
/// <summary>
@ -191,11 +196,7 @@ namespace GameCult.Mesh
var completion = new TaskCompletionSource<CultMeshVerseCatalogResponseMessage>(
TaskCreationOptions.RunContinuationsAsynchronously);
using var client = new Client(_options.Security ?? ClientSecurityOptions.Development())
{
AllowUnverifiedCultNetMessages = true
};
_options.ConfigureClient?.Invoke(client);
using var client = CreateClient();
client.OnCultNet<CultMeshVerseCatalogResponseMessage>(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)

99
src/GameCult.Networking/CultNetSchemaClient.cs Normal file
View file

@ -0,0 +1,99 @@
using System;
namespace GameCult.Networking
{
/// <summary>
/// Minimal client port for request/response services that speak CultNet schema-v0 messages.
/// </summary>
public interface ICultNetSchemaClient : IDisposable
{
/// <summary>
/// Gets whether the underlying transport reports an established connection.
/// </summary>
bool Connected { get; }
/// <summary>
/// Connects the transport to a remote endpoint.
/// </summary>
void Connect(string host, int port);
/// <summary>
/// Sends one schema-v0 message through the transport.
/// </summary>
void SendCultNet<T>(T message)
where T : ICultNetSchemaMessage;
/// <summary>
/// Adds a callback for one schema-v0 response type.
/// </summary>
void OnCultNet<T>(Action<T> callback)
where T : ICultNetSchemaMessage;
}
/// <summary>
/// LiteNetLib-backed implementation of the CultNet schema client port.
/// </summary>
public sealed class LiteNetLibCultNetSchemaClient : ICultNetSchemaClient
{
private readonly Client _client;
/// <summary>
/// Creates a LiteNetLib schema client adapter.
/// </summary>
public LiteNetLibCultNetSchemaClient(
ClientSecurityOptions? security = null,
Action<Client>? configureClient = null)
{
_client = new Client(security ?? ClientSecurityOptions.Development())
{
AllowUnverifiedCultNetMessages = true
};
configureClient?.Invoke(_client);
}
/// <inheritdoc />
public bool Connected => _client.Connected;
/// <inheritdoc />
public void Connect(string host, int port)
{
_client.Connect(host, port);
}
/// <inheritdoc />
public void SendCultNet<T>(T message)
where T : ICultNetSchemaMessage
{
_client.SendCultNet(message);
}
/// <inheritdoc />
public void OnCultNet<T>(Action<T> callback)
where T : ICultNetSchemaMessage
{
_client.OnCultNet(callback);
}
/// <inheritdoc />
public void Dispose()
{
_client.Dispose();
}
}
/// <summary>
/// Factory helpers for schema client adapters.
/// </summary>
public static class CultNetSchemaClients
{
/// <summary>
/// Creates the default C# LiteNetLib schema client adapter.
/// </summary>
public static ICultNetSchemaClient CreateLiteNetLib(
ClientSecurityOptions? security = null,
Action<Client>? configureClient = null)
{
return new LiteNetLibCultNetSchemaClient(security, configureClient);
}
}
}

38
src/GameCult.Networking/CultNetShardReplication.cs
View file

@ -393,6 +393,11 @@ namespace GameCult.Networking
/// Gets or sets a callback used to customize each ephemeral fetch client.
/// </summary>
public Action<Client>? ConfigureClient { get; set; }
/// <summary>
/// Gets or sets the schema-v0 client factory. Defaults to the C# LiteNetLib adapter.
/// </summary>
public Func<ICultNetSchemaClient>? CreateClient { get; set; }
}
/// <summary>
@ -423,11 +428,7 @@ namespace GameCult.Networking
var completion = new TaskCompletionSource<CultNetShardLogResponseMessage>(
TaskCreationOptions.RunContinuationsAsynchronously);
using var client = new Client(_options.Security ?? ClientSecurityOptions.Development())
{
AllowUnverifiedCultNetMessages = true
};
_options.ConfigureClient?.Invoke(client);
using var client = CreateClient();
client.OnCultNet<CultNetShardLogResponseMessage>(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.
/// </summary>
public Action<Client>? ConfigureClient { get; set; }
/// <summary>
/// Gets or sets the schema-v0 client factory. Defaults to the C# LiteNetLib adapter.
/// </summary>
public Func<ICultNetSchemaClient>? CreateClient { get; set; }
}
/// <summary>
@ -539,11 +551,7 @@ namespace GameCult.Networking
var completion = new TaskCompletionSource<CultNetSnapshotResponseRawMessage>(
TaskCreationOptions.RunContinuationsAsynchronously);
using var client = new Client(_options.Security ?? ClientSecurityOptions.Development())
{
AllowUnverifiedCultNetMessages = true
};
_options.ConfigureClient?.Invoke(client);
using var client = CreateClient();
client.OnCultNet<CultNetSnapshotResponseRawMessage>(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)

19
src/GameCult.Networking/CultNetShardWriteForwarding.cs
View file

@ -55,6 +55,11 @@ namespace GameCult.Networking
/// Gets or sets a callback used to customize each ephemeral forwarding client.
/// </summary>
public Action<Client>? ConfigureClient { get; set; }
/// <summary>
/// Gets or sets the schema-v0 client factory. Defaults to the C# LiteNetLib adapter.
/// </summary>
public Func<ICultNetSchemaClient>? CreateClient { get; set; }
}
/// <summary>
@ -115,17 +120,19 @@ namespace GameCult.Networking
private async Task SendAsync<T>(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)

203
tests/GameCult.Networking.Tests/NetworkingTests.cs
View file

@ -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<CultMeshVerseCatalogRequestMessage>());
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<CultMeshPeerExchangeRequestMessage>());
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<CultNetShardLogRequestMessage>());
Assert.That(log.ShardId, Is.EqualTo("notes"));
Assert.That(snapshotClient.SentMessages.Single(), Is.TypeOf<CultNetSnapshotRequestMessage>());
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<ICultNetSchemaMessage, ICultNetSchemaMessage?> _respond;
private readonly Dictionary<Type, List<Delegate>> _handlers = new Dictionary<Type, List<Delegate>>();
public CapturingSchemaClient(Func<ICultNetSchemaMessage, ICultNetSchemaMessage?> respond)
{
_respond = respond;
}
public bool Connected { get; private set; }
public string? ConnectedHost { get; private set; }
public int ConnectedPort { get; private set; }
public List<ICultNetSchemaMessage> SentMessages { get; } = new List<ICultNetSchemaMessage>();
public void Connect(string host, int port)
{
Connected = true;
ConnectedHost = host;
ConnectedPort = port;
}
public void SendCultNet<T>(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<T>(Action<T> callback)
where T : ICultNetSchemaMessage
{
if (!_handlers.TryGetValue(typeof(T), out var handlers))
{
handlers = new List<Delegate>();
_handlers[typeof(T)] = handlers;
}
handlers.Add(callback);
}
public void Dispose()
{
}
}
[MessagePack.MessagePackObject]
public sealed class PlayerDataPayload
{