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

Stream exact hot state over negotiated body planes

Browse source
This commit is contained in:
Meta Vens 2026-07-19 05:54:24 +01:00
commit 743fa3889e
22 changed files with 320 additions and 14 deletions
Download patch file Download diff file Expand all files Collapse all files

2
Directory.Build.props
View file

@ -14,7 +14,7 @@
<PackageLicenseExpression>MIT</PackageLicenseExpression>
<Copyright>© Meta Vens and contributors</Copyright>
<PublishRepositoryUrl>true</PublishRepositoryUrl>
<Version>1.0.22</Version>
<Version>1.0.23</Version>
<!-- Symbols/debugging -->
<EmbedUntrackedSources>true</EmbedUntrackedSources>

34
src/GameCult.Mesh/CultMesh.cs
View file

@ -2852,6 +2852,40 @@ namespace GameCult.Mesh
return new CultMeshRudpEndpoint(host, uri.Port, $"rudp://{uriHost}:{uri.Port}");
}
/// <summary>
/// Opens one exact hot-body subscription and advertises every body plane this runtime can
/// actually open. The provider derives locality and publishes only the selected plane.
/// </summary>
public static Task<IReadOnlyList<object>> SubscribeHotBodyAsync(
CultNetDatabaseSubscriptionClient subscriptions,
IEnumerable<CultMeshBodyTransportKind> supportedBodyTransports,
CultMeshHotBodySubscription subscription)
{
if (subscriptions == null) throw new ArgumentNullException(nameof(subscriptions));
if (supportedBodyTransports == null) throw new ArgumentNullException(nameof(supportedBodyTransports));
if (subscription == null) throw new ArgumentNullException(nameof(subscription));
var transports = supportedBodyTransports.Distinct().ToArray();
if (transports.Length == 0)
throw new ArgumentException("At least one readable body transport is required.", nameof(supportedBodyTransports));
return subscriptions.SubscribeAsync(
subscription.SubscriptionId,
recordKeys: subscription.RecordKeys,
schemaIds: subscription.SchemaIds,
consumerRuntimeId: subscription.ConsumerRuntimeId,
bodyIds: new[] { subscription.BodyId },
supportedBodyTransports: transports.Select(value => value.ToString()));
}
/// <summary>Opens an exact hot-body subscription using a resolver's readable planes.</summary>
public static Task<IReadOnlyList<object>> SubscribeHotBodyAsync(
CultNetDatabaseSubscriptionClient subscriptions,
CultMeshBodyPublicationResolver bodyResolver,
CultMeshHotBodySubscription subscription)
{
if (bodyResolver == null) throw new ArgumentNullException(nameof(bodyResolver));
return SubscribeHotBodyAsync(subscriptions, bodyResolver.SupportedTransports, subscription);
}
/// <summary>Resolves a CultMesh URI or RUDP endpoint into a concrete RUDP endpoint.</summary>
public static CultMeshRudpEndpoint ResolveRudpEndpoint(string endpointOrUri)
{

12
src/GameCult.Mesh/CultMeshBodies.cs
View file

@ -150,6 +150,11 @@ namespace GameCult.Mesh
private readonly Func<CultMeshBodyDescriptor, bool> _authorizeProducer;
private readonly Func<string, CultMeshBodyDescriptor, bool>? _authorizeNamedProducer;
/// <summary>Gets locally available body planes in fastest-local-first order.</summary>
public IReadOnlyList<CultMeshBodyTransportKind> SupportedTransports => _adapters.Keys
.OrderByDescending(TransportPreference)
.ToArray();
public CultMeshBodyTransportService(
IEnumerable<ICultMeshBodyTransportAdapter> adapters,
Func<CultMeshBodyDescriptor, bool> authorizeProducer)
@ -264,6 +269,13 @@ namespace GameCult.Mesh
left.AccessMode == right.AccessMode &&
left.Synchronization == right.Synchronization &&
string.Equals(left.SemanticHash, right.SemanticHash, StringComparison.Ordinal);
private static int TransportPreference(CultMeshBodyTransportKind transport) => transport switch
{
CultMeshBodyTransportKind.SharedMemory => 3,
CultMeshBodyTransportKind.SharedFileMapping => 2,
_ => 1
};
}
public sealed class CultMeshMappedBodyPublisher

51
src/GameCult.Mesh/CultMeshBodyDemand.cs
View file

@ -7,6 +7,57 @@ using GameCult.Networking;
namespace GameCult.Mesh
{
/// <summary>
/// One exact hot-body subscription. The view record, latest-publication record, consumer body
/// demand, and locally supported planes are one contract so callers cannot accidentally stream
/// body bytes through retained document snapshots.
/// </summary>
public sealed class CultMeshHotBodySubscription
{
public CultMeshHotBodySubscription(
string subscriptionId,
string consumerRuntimeId,
string viewRecordKey,
string viewSchemaId,
string bodyId,
IEnumerable<string>? additionalRecordKeys = null,
IEnumerable<string>? additionalSchemaIds = null)
{
SubscriptionId = Require(subscriptionId, nameof(subscriptionId));
ConsumerRuntimeId = Require(consumerRuntimeId, nameof(consumerRuntimeId));
BodyId = Require(bodyId, nameof(bodyId));
RecordKeys = new[]
{
Require(viewRecordKey, nameof(viewRecordKey)),
CultMeshBodyPublicationDocument.CreateLatestRecordKey(bodyId).Value
}
.Concat(additionalRecordKeys ?? Array.Empty<string>())
.Where(value => !string.IsNullOrWhiteSpace(value))
.Distinct(StringComparer.Ordinal)
.ToArray();
SchemaIds = new[]
{
Require(viewSchemaId, nameof(viewSchemaId)),
CultMeshBodyPublicationSchemaVersions.Publication
}
.Concat(additionalSchemaIds ?? Array.Empty<string>())
.Where(value => !string.IsNullOrWhiteSpace(value))
.Distinct(StringComparer.Ordinal)
.ToArray();
}
public string SubscriptionId { get; }
public string ConsumerRuntimeId { get; }
public string BodyId { get; }
public IReadOnlyList<string> RecordKeys { get; }
public IReadOnlyList<string> SchemaIds { get; }
private static string Require(string value, string parameterName) =>
string.IsNullOrWhiteSpace(value)
? throw new ArgumentException("Value is required.", parameterName)
: value;
}
/// <summary>Derived transport work for one logical hot body.</summary>
public sealed class CultMeshBodyDemandPlan
{

4
src/GameCult.Mesh/CultMeshBodyPublications.cs
View file

@ -1,5 +1,6 @@
using System;
using System.Globalization;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Security.Cryptography;
@ -361,6 +362,9 @@ namespace GameCult.Mesh
public CultMeshBodyPublicationResolver(CultMeshBodyTransportService transport) =>
_transport = transport ?? throw new ArgumentNullException(nameof(transport));
/// <summary>Gets the body planes this resolver can actually open.</summary>
public IReadOnlyList<CultMeshBodyTransportKind> SupportedTransports => _transport.SupportedTransports;
public ICultMeshBodyReadLease ResolveReadOnly(
CultMeshBodyPublicationDocument publication,
CultMeshBodyValidationRequest request) =>

14
src/GameCult.Mesh/docs/reliability-migration-plan.md
View file

@ -918,6 +918,14 @@ stream demand merely by existing.
and server disposal all update the same typed demand projection. Stable peer
identity, not per-message wrapper identity, keys subscription lifetime.
`CultMesh.SubscribeHotBodyAsync(...)` is the public construction path. Its
`CultMeshHotBodySubscription` keeps the exact view record, the body's stable
latest-publication record, consumer identity, and logical body demand in one
contract. The local `CultMeshBodyTransportService` contributes only adapters it
can actually open, ordered shared-memory first; the server combines those
capabilities with observed peer locality. Applications do not hand-assemble
transport-name arrays or request one snapshot per frame.
**Cut line:** local frame publication writes directly into a mapped write lease
and commits its descriptor. The compatibility byte-array publisher is copy-in
only. The current remote frame representation may still use the verified CDN
@ -925,6 +933,12 @@ body path when remote demand exists; Phase 6 must replace that expiry-sensitive
traffic with a stream/datagram body plane rather than enlarging snapshot or CDN
timeouts.
Reliable control messages larger than the RUDP receive mask explicitly
acknowledge the packet just received. A retransmit older than the rolling
32-packet summary can therefore always leave the sender queue; the rolling mask
remains useful for piggyback acknowledgements but is not allowed to strand a
fragment indefinitely.
### Gate
- Two local consumers map one verified CDN artifact without duplicate process

27
src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs
View file

@ -163,16 +163,17 @@ namespace GameCult.Networking
{
var changeType = change.GetType();
var key = (CultRecordKey)(changeType.GetProperty("Key")?.GetValue(change) ?? new CultRecordKey(""));
var schemaId = (string?)changeType.GetProperty("SchemaId")?.GetValue(change) ?? "";
if (request.RecordKeys is { Length: > 0 } && !request.RecordKeys.Contains(key.Value, StringComparer.Ordinal))
return null;
if (request.SchemaIds is { Length: > 0 } && !request.SchemaIds.Contains(schemaId, StringComparer.Ordinal))
return null;
var kind = (CultNetDatabaseChangeKind)(changeType.GetProperty("Kind")?.GetValue(change) ?? CultNetDatabaseChangeKind.Updated);
var document = changeType.GetProperty("Document")?.GetValue(change);
if (kind == CultNetDatabaseChangeKind.Removed || document == null)
{
var previous = changeType.GetProperty("PreviousDocument")?.GetValue(change);
var schemaId = ResolveWireSchemaId(previous, (string?)changeType.GetProperty("SchemaId")?.GetValue(change) ?? "");
if (!MatchesRequestedSchema(request.SchemaIds, previous, schemaId))
return null;
return new CultNetDatabaseChangeRawMessage
{
MessageId = Guid.NewGuid().ToString("N"),
@ -183,15 +184,33 @@ namespace GameCult.Networking
};
}
var raw = CreateRawRecord(key, document);
if (!MatchesRequestedSchema(request.SchemaIds, document, raw.SchemaId))
return null;
return new CultNetDatabaseChangeRawMessage
{
MessageId = Guid.NewGuid().ToString("N"),
SubscriptionId = subscriptionId,
ChangeKind = kind == CultNetDatabaseChangeKind.Added ? "added" : "updated",
Document = CreateRawRecord(key, document)
Document = raw
};
}
private string ResolveWireSchemaId(object? document, string fallback)
{
if (document == null) return fallback;
return _database.Documents.GetByDocumentType(document.GetType())?.SchemaId ?? fallback;
}
private bool MatchesRequestedSchema(string[]? requestedSchemaIds, object? document, string wireSchemaId)
{
if (requestedSchemaIds is not { Length: > 0 }) return true;
if (requestedSchemaIds.Contains(wireSchemaId, StringComparer.Ordinal)) return true;
return document != null && CultNetSchemaAliasMatching.MatchesAny(
requestedSchemaIds,
_database.Cache.Registry.GetRequired(document.GetType()));
}
private CultNetRawDocumentRecord CreateRawRecord(CultRecordKey key, object document)
{
var method = typeof(CultNetDocumentRegistry)

20
src/GameCult.Networking/CultNetTransport.cs
View file

@ -1141,9 +1141,14 @@ namespace GameCult.Networking
/// <summary>
/// Creates a packet carrying the current acknowledgement state.
/// </summary>
public CultNetRudpPacket CreateAck()
public CultNetRudpPacket CreateAck(uint? acknowledgedSequence = null)
{
var (ack, ackMask) = AckState();
if (acknowledgedSequence.HasValue)
{
ack = acknowledgedSequence.Value;
ackMask = 0;
}
return new CultNetRudpPacket
{
PacketType = CultNetRudpPacketType.Ack,
@ -1955,10 +1960,9 @@ namespace GameCult.Networking
delivered = _deliveredFrames.Count > 0 ? _deliveredFrames.Dequeue() : null;
if (packet.PacketType == CultNetRudpPacketType.Accept ||
packet.PacketType == CultNetRudpPacketType.Data ||
delivered != null)
packet.PacketType == CultNetRudpPacketType.Data)
{
SendPacket(_session.CreateAck());
SendPacket(_session.CreateAck(packet.Sequence));
}
return true;
@ -2108,6 +2112,10 @@ namespace GameCult.Networking
/// </summary>
public bool Connected => Session.Connected;
/// <summary>
/// Gets the number of reliable packets still awaiting acknowledgement.
/// </summary>
public int PendingReliablePacketCount => Session.PendingReliableSequences.Count;
/// <summary>
/// Gets the last transport-level remote disconnect reason, if one was received.
/// </summary>
public byte[]? DisconnectReason { get; internal set; }
@ -2349,9 +2357,9 @@ namespace GameCult.Networking
}
delivered = _deliveredFrames.Count > 0 ? _deliveredFrames.Dequeue() : null;
if (packet.PacketType == CultNetRudpPacketType.Data || delivered != null)
if (packet.PacketType == CultNetRudpPacketType.Data)
{
SendPacket(existingPeer.RemoteEndPoint, existingPeer.Session.CreateAck());
SendPacket(existingPeer.RemoteEndPoint, existingPeer.Session.CreateAck(packet.Sequence));
}
return true;

24
tests/GameCult.Mesh.Tests/CultMeshBodyTransportTests.cs
View file

@ -357,6 +357,30 @@ public sealed class CultMeshBodyTransportTests
plan.Consumers.Should().ContainSingle().Which.ConsumerRuntimeId.Should().Be("unknown");
}
[Test]
public void HotBodySubscriptionKeepsViewPublicationAndDemandInOneExactContract()
{
var subscription = new CultMeshHotBodySubscription(
"pilot-world",
"eve-unity",
"eve:view:pilot",
"gamecult.eve.entity_soa_view.v2",
"eve:body:pilot",
new[] { "eve:field:gravity" },
new[] { "gamecult.eve.fields_splats.v1" });
subscription.RecordKeys.Should().Equal(
"eve:view:pilot",
CultMeshBodyPublicationDocument.CreateLatestRecordKey("eve:body:pilot").Value,
"eve:field:gravity");
subscription.SchemaIds.Should().Equal(
"gamecult.eve.entity_soa_view.v2",
CultMeshBodyPublicationSchemaVersions.Publication,
"gamecult.eve.fields_splats.v1");
subscription.BodyId.Should().Be("eve:body:pilot");
subscription.ConsumerRuntimeId.Should().Be("eve-unity");
}
private static CultMeshBodyValidationRequest Request(CultMeshBodyDescriptor descriptor, DateTimeOffset now) => new()
{
BodyId = descriptor.BodyId,

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

@ -771,6 +771,43 @@ namespace GameCult.Networking.Tests
Assert.That(session.DueResends(250), Is.Empty);
}
[Test]
public void RudpSession_ExplicitlyAcknowledgesRetransmitsOlderThanTheReceiveMask()
{
var sender = new CultNetRudpSession(new CultNetRudpSessionOptions
{
ConnectionId = 991,
InitialSequence = 1,
ResendDelayMs = 10
});
var receiver = new CultNetRudpSession(new CultNetRudpSessionOptions
{
ConnectionId = 991,
InitialSequence = 500
});
sender.Receive(new CultNetRudpPacket
{ PacketType = CultNetRudpPacketType.Accept, ConnectionId = 991, Sequence = 400, ChannelId = "control" });
receiver.Receive(new CultNetRudpPacket
{ PacketType = CultNetRudpPacketType.Accept, ConnectionId = 991, Sequence = 0, ChannelId = "control" });
var packets = sender.SendMany(
"snapshot",
Enumerable.Range(0, 80).Select(index => (byte)index).ToArray(),
new CultNetRudpSendOptions { Reliable = true, Ordered = true, NowMs = 0 },
maxFragmentBytes: 1);
foreach (var packet in packets)
receiver.Receive(packet);
sender.Receive(receiver.CreateAck());
Assert.That(sender.PendingReliableSequences.First(), Is.EqualTo(1u));
var oldRetransmit = sender.DueResends(10).Single(packet => packet.Sequence == 1);
receiver.Receive(oldRetransmit);
sender.Receive(receiver.CreateAck(oldRetransmit.Sequence));
Assert.That(sender.PendingReliableSequences, Does.Not.Contain(1u));
}
[Test]
public async Task RudpSession_ResendsAndAcknowledgementsCanRunConcurrently()
{
@ -1544,6 +1581,104 @@ namespace GameCult.Networking.Tests
Assert.That(((NetworkSchemaNote)targetCache.Get(new CultRecordKey(recordKey))!).Text, Is.EqualTo("live"));
}
[Test]
public async Task DatabaseSubscription_FiltersLiveChangesByWireSchemaBinding()
{
const string recordKey = "tests:subscription-client:wire-note";
const string wireSchema = "tests.wire_note.v1";
var sourceCache = new CultCache();
var sourceDocuments = new CultNetDocumentRegistry(sourceCache.Registry)
.Register(CultNetDocumentBinding.ForDocument<NetworkSchemaNote>(sourceCache.Registry, wireSchema));
using var sourceDatabase = new CultNetDatabase(sourceCache, new CultNetDatabaseOptions
{
DocumentRegistry = sourceDocuments
});
using var server = new RudpCultNetSchemaServer(new RudpCultNetSchemaServerOptions
{
RuntimeId = "database-wire-schema-subscription-server",
Socket = BindUdpSocket()
});
using var subscriptions = new CultNetDatabaseSubscriptionServer(server, sourceDatabase);
using var cancellation = new CancellationTokenSource();
var serverThread = new Thread(() =>
{
while (!cancellation.IsCancellationRequested)
{
_ = server.PollOnceAsync().GetAwaiter().GetResult();
Thread.Sleep(1);
}
}) { IsBackground = true };
serverThread.Start();
var targetCache = new CultCache();
var targetDocuments = new CultNetDocumentRegistry(targetCache.Registry)
.Register(CultNetDocumentBinding.ForDocument<NetworkSchemaNote>(targetCache.Registry, wireSchema));
var transport = CultNetSchemaClients.CreateRudp("database-wire-schema-subscription-client");
using var client = new CultNetDatabaseSubscriptionClient(transport, targetCache, targetDocuments);
transport.Connect("127.0.0.1", server.LocalEndPoint.Port);
await WaitUntilAsync(() => transport.Connected, TimeSpan.FromSeconds(2));
await AwaitWithTimeout(
client.SubscribeAsync("wire-notes", recordKeys: [recordKey], schemaIds: [wireSchema]),
TimeSpan.FromSeconds(2));
var changed = new TaskCompletionSource<CultNetReplicatedDocumentChange>(TaskCreationOptions.RunContinuationsAsynchronously);
client.Changed += change => changed.TrySetResult(change);
await sourceDatabase.PutAsync(new CultRecordKey(recordKey), new NetworkSchemaNote
{
Schema = "tests.networking_note.v1",
Text = "wire-live"
});
var update = await AwaitWithTimeout(changed.Task, TimeSpan.FromSeconds(2));
cancellation.Cancel();
Assert.That(update.SchemaId, Is.EqualTo(wireSchema));
Assert.That(((NetworkSchemaNote)update.Document!).Text, Is.EqualTo("wire-live"));
}
[Test]
public void DatabaseSubscription_FiltersCanonicalWireChangesByRequestedSchemaAlias()
{
var cache = new CultCache();
var documents = new CultNetDocumentRegistry(cache.Registry)
.Register(CultNetDocumentBinding.ForDocument<NetworkSchemaNote>(cache.Registry));
using var database = new CultNetDatabase(cache, new CultNetDatabaseOptions
{
DocumentRegistry = documents
});
using var server = new RudpCultNetSchemaServer(new RudpCultNetSchemaServerOptions
{
RuntimeId = "database-schema-alias-subscription-server",
Socket = BindUdpSocket()
});
using var subscriptions = new CultNetDatabaseSubscriptionServer(server, database);
var descriptor = cache.Registry.GetRequired<NetworkSchemaNote>();
var change = new CultNetDatabaseChange<NetworkSchemaNote>(
CultNetDatabaseChangeKind.Added,
new CultRecordKey("tests:subscription-client:alias-note"),
descriptor.SchemaId,
database.Shards[0],
new NetworkSchemaNote { Schema = "tests.networking_note.v1", Text = "alias-live" },
previousDocument: null);
var method = typeof(CultNetDatabaseSubscriptionServer).GetMethod(
"CreateChange",
BindingFlags.Instance | BindingFlags.NonPublic)!;
var message = (CultNetDatabaseChangeRawMessage?)method.Invoke(subscriptions, new object[]
{
change,
new CultNetDatabaseSubscribeMessage
{
SubscriptionId = "alias-notes",
SchemaIds = ["tests.networking_note.v1"],
RecordKeys = ["tests:subscription-client:alias-note"]
},
"alias-notes"
});
Assert.That(message, Is.Not.Null);
Assert.That(message!.Document!.SchemaId, Is.EqualTo(descriptor.SchemaId));
}
[Test]
public async Task RudpCultNetSchemaServer_DeliversLargeSnapshotResponse()
{
@ -1607,9 +1742,14 @@ namespace GameCult.Networking.Tests
if (completed != responseCompletion.Task)
Assert.Fail($"Large snapshot timed out; server sent {server.Stats.BytesSent} bytes, received {server.Stats.BytesReceived} bytes, transports {server.Profile.Transports.Count()}.");
var response = await AwaitWithTimeout(responseCompletion.Task, TimeSpan.FromMilliseconds(1));
await WaitUntilAsync(
() => server.Peers.Count == 1 && server.Peers.Single().PendingReliablePacketCount == 0,
TimeSpan.FromSeconds(2));
cancellation.Cancel();
Assert.That(response.Documents, Has.Length.EqualTo(1));
Assert.That(response.Documents[0].Payload, Has.Length.EqualTo(128 * 1024));
Assert.That(server.Stats.BytesSent, Is.LessThan(512 * 1024),
"A lossless large response must not remain trapped in the reliable resend queue.");
}
[Test]

4
unity/org.gamecult.cultlib/README.md
View file

@ -7,8 +7,8 @@ CultCache, CultNet, and CultMesh assemblies consumed through NuGet by normal
Build it with `scripts/build-unity-package.ps1`. Do not edit generated plugin
assemblies in the package output.
Install version `1.0.22` through Unity Package Manager with:
Install version `1.0.23` through Unity Package Manager with:
```text
https://github.com/GameCult/CultLib.git?path=/unity/org.gamecult.cultlib#cultlib-unity-v1.0.22
https://github.com/GameCult/CultLib.git?path=/unity/org.gamecult.cultlib#cultlib-unity-v1.0.23
```

BIN
unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.MessagePack.dll
View file

Binary file not shown.

BIN
unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.MessagePack.pdb
View file

Binary file not shown.

BIN
unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.dll
View file

Binary file not shown.

BIN
unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.pdb
View file

Binary file not shown.

BIN
unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Logging.dll
View file

Binary file not shown.

BIN
unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Logging.pdb
View file

Binary file not shown.

BIN
unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.dll
View file

Binary file not shown.

BIN
unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.pdb
View file

Binary file not shown.

BIN
unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.dll
View file

Binary file not shown.

BIN
unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.pdb
View file

Binary file not shown.

2
unity/org.gamecult.cultlib/package.json
View file

@ -1,7 +1,7 @@
{
"name": "org.gamecult.cultlib",
"displayName": "CultLib",
"version": "1.0.22",
"version": "1.0.23",
"unity": "2021.3",
"description": "CultCache, CultNet, and CultMesh managed runtime assemblies for Unity.",
"license": "MIT",