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

Add ergonomic live body subscriptions

Browse source
This commit is contained in:
Meta Vens 2026-07-19 10:55:27 +01:00
commit 5587f5c6d1
19 changed files with 382 additions and 24 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.27</Version>
<Version>1.0.28</Version>
<!-- Symbols/debugging -->
<EmbedUntrackedSources>true</EmbedUntrackedSources>

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

@ -5,6 +5,7 @@ using System.Linq;
using System.Net;
using System.Net.Sockets;
using System.Reflection;
using System.Threading;
using System.Threading.Tasks;
using GameCult.Caching;
using GameCult.Caching.MessagePack;
@ -2889,6 +2890,51 @@ namespace GameCult.Mesh
return SubscribeHotBodyAsync(subscriptions, bodyResolver.SupportedTransports, subscription, deliveryMode);
}
/// <summary>
/// Subscribes to one logical body's latest descriptor as ephemeral control state and declares
/// demand for the body itself. Body bytes are opened separately through the fastest valid plane.
/// </summary>
public static async Task<CultMeshLiveBody> SubscribeLiveBodyAsync(
CultNetDatabaseSubscriptionClient subscriptions,
CultMeshBodyPublicationResolver bodyResolver,
CultMeshLiveBodySubscription subscription,
CancellationToken cancellationToken = default)
{
if (subscriptions == null) throw new ArgumentNullException(nameof(subscriptions));
if (bodyResolver == null) throw new ArgumentNullException(nameof(bodyResolver));
if (subscription == null) throw new ArgumentNullException(nameof(subscription));
var transports = bodyResolver.SupportedTransports.Distinct().ToArray();
if (transports.Length == 0)
throw new InvalidOperationException("The body resolver has no readable transport planes.");
var liveBody = new CultMeshLiveBody(subscriptions, bodyResolver, subscription);
try
{
var initial = await subscriptions.SubscribeAsync(
subscription.SubscriptionId,
recordKeys: new[] { subscription.PublicationRecordKey },
schemaIds: new[] { CultMeshBodyPublicationSchemaVersions.Publication },
consumerRuntimeId: subscription.ConsumerRuntimeId,
bodyIds: new[] { subscription.BodyId },
supportedBodyTransports: transports.Select(value => value.ToString()),
deliveryMode: CultNetDatabaseSubscriptionDeliveryMode.Live,
cancellationToken: cancellationToken)
.ConfigureAwait(false);
var publications = initial.OfType<CultMeshBodyPublicationDocument>().ToArray();
if (publications.Length != 1)
throw new InvalidOperationException(
$"Live body subscription '{subscription.SubscriptionId}' expected one publication for " +
$"'{subscription.BodyId}' but received {publications.Length}.");
liveBody.Initialize(publications[0]);
return liveBody;
}
catch
{
liveBody.Dispose();
throw;
}
}
/// <summary>Resolves a CultMesh URI or RUDP endpoint into a concrete RUDP endpoint.</summary>
public static CultMeshRudpEndpoint ResolveRudpEndpoint(string endpointOrUri)
{

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

@ -7,6 +7,169 @@ using GameCult.Networking;
namespace GameCult.Mesh
{
/// <summary>
/// Declares one logical body demand. CultMesh owns locality detection and transport selection;
/// consumers name the body and the planes they can actually read.
/// </summary>
public sealed class CultMeshLiveBodySubscription
{
public CultMeshLiveBodySubscription(
string subscriptionId,
string consumerRuntimeId,
string bodyId)
{
SubscriptionId = Require(subscriptionId, nameof(subscriptionId));
ConsumerRuntimeId = Require(consumerRuntimeId, nameof(consumerRuntimeId));
BodyId = Require(bodyId, nameof(bodyId));
}
public string SubscriptionId { get; }
public string ConsumerRuntimeId { get; }
public string BodyId { get; }
public string PublicationRecordKey =>
CultMeshBodyPublicationDocument.CreateLatestRecordKey(BodyId).Value;
private static string Require(string value, string parameterName) =>
string.IsNullOrWhiteSpace(value)
? throw new ArgumentException("Value is required.", parameterName)
: value;
}
/// <summary>
/// One ephemeral body subscription. Publication metadata remains reactive control state while
/// body bytes stay on the negotiated body plane.
/// </summary>
public sealed class CultMeshLiveBody : IDisposable
{
private readonly CultNetDatabaseSubscriptionClient _owner;
private readonly CultMeshBodyPublicationResolver _resolver;
private readonly CultMeshLiveBodySubscription _subscription;
private readonly object _gate = new();
private CultMeshBodyPublicationDocument? _current;
private bool _hasValue;
private bool _disposed;
internal CultMeshLiveBody(
CultNetDatabaseSubscriptionClient owner,
CultMeshBodyPublicationResolver resolver,
CultMeshLiveBodySubscription subscription)
{
_owner = owner ?? throw new ArgumentNullException(nameof(owner));
_resolver = resolver ?? throw new ArgumentNullException(nameof(resolver));
_subscription = subscription ?? throw new ArgumentNullException(nameof(subscription));
_owner.Changed += OnChanged;
}
/// <summary>Raised when the provider publishes a new generation descriptor.</summary>
public event Action<CultMeshBodyPublicationDocument>? Changed;
/// <summary>Raised when the provider withdraws the logical body.</summary>
public event Action? Removed;
public string BodyId => _subscription.BodyId;
public bool HasValue
{
get { lock (_gate) return _hasValue; }
}
public CultMeshBodyPublicationDocument Current
{
get
{
lock (_gate)
return _hasValue
? _current!
: throw new InvalidOperationException(
$"Live body '{_subscription.BodyId}' is not currently published.");
}
}
/// <summary>
/// Opens the current generation read-only using the fastest valid advertised representation.
/// The caller owns the returned lease and should dispose it before opening a later generation.
/// </summary>
public CultMeshBodyNegotiationResult OpenCurrentReadOnly(DateTimeOffset? nowUtc = null)
{
CultMeshBodyPublicationDocument publication;
lock (_gate)
{
if (_disposed) throw new ObjectDisposedException(nameof(CultMeshLiveBody));
if (!_hasValue)
throw new InvalidOperationException(
$"Live body '{_subscription.BodyId}' is not currently published.");
publication = _current!;
}
return _resolver.NegotiateReadOnly(publication, new CultMeshBodyValidationRequest
{
BodyId = publication.BodyId,
SchemaId = publication.SchemaId,
LayoutVersion = publication.LayoutVersion,
ProducerEpoch = publication.ProducerEpoch,
Sequence = publication.Sequence,
Capacity = publication.Capacity,
AccessMode = CultMeshBodyAccessMode.ReadOnly,
NowUtc = nowUtc ?? DateTimeOffset.UtcNow
});
}
internal void Initialize(CultMeshBodyPublicationDocument publication)
{
CultMeshBodyPublicationValidator.Validate(publication, expectedBodyId: _subscription.BodyId);
lock (_gate)
{
if (_disposed || _hasValue) return;
_current = publication;
_hasValue = true;
}
}
public void Dispose()
{
lock (_gate)
{
if (_disposed) return;
_disposed = true;
_current = null;
_hasValue = false;
}
_owner.Changed -= OnChanged;
_owner.Unsubscribe(_subscription.SubscriptionId);
}
private void OnChanged(CultNetReplicatedDocumentChange change)
{
if (!string.Equals(change.SubscriptionId, _subscription.SubscriptionId, StringComparison.Ordinal) ||
!string.Equals(change.RecordKey, _subscription.PublicationRecordKey, StringComparison.Ordinal) ||
!string.Equals(change.SchemaId, CultMeshBodyPublicationSchemaVersions.Publication, StringComparison.Ordinal))
return;
if (string.Equals(change.ChangeKind, "removed", StringComparison.Ordinal))
{
lock (_gate)
{
if (_disposed) return;
_current = null;
_hasValue = false;
}
Removed?.Invoke();
return;
}
if (change.Document is not CultMeshBodyPublicationDocument publication)
return;
CultMeshBodyPublicationValidator.Validate(publication, expectedBodyId: _subscription.BodyId);
lock (_gate)
{
if (_disposed) return;
_current = publication;
_hasValue = true;
}
Changed?.Invoke(publication);
}
}
/// <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
@ -68,12 +231,14 @@ namespace GameCult.Mesh
BodyId = bodyId;
Consumers = consumers;
RequiresSharedMemory = consumers.Any(value => value.Transport == CultMeshBodyTransportKind.SharedMemory);
RequiresSharedFileMapping = consumers.Any(value => value.Transport == CultMeshBodyTransportKind.SharedFileMapping);
RequiresNetwork = consumers.Any(value => value.Transport == CultMeshBodyTransportKind.Network);
}
public string BodyId { get; }
public bool HasConsumers => Consumers.Count > 0;
public bool RequiresSharedMemory { get; }
public bool RequiresSharedFileMapping { get; }
public bool RequiresNetwork { get; }
public IReadOnlyList<CultMeshBodyConsumerRoute> Consumers { get; }
}
@ -172,6 +337,11 @@ namespace GameCult.Mesh
demand.ConsumerRuntimeId,
demand.SubscriptionId,
CultMeshBodyTransportKind.SharedMemory);
if (demand.SameMachine && Supports(supported, CultMeshBodyTransportKind.SharedFileMapping))
return new CultMeshBodyConsumerRoute(
demand.ConsumerRuntimeId,
demand.SubscriptionId,
CultMeshBodyTransportKind.SharedFileMapping);
if (Supports(supported, CultMeshBodyTransportKind.Network))
return new CultMeshBodyConsumerRoute(
demand.ConsumerRuntimeId,

37
src/GameCult.Mesh/README.md
View file

@ -389,28 +389,29 @@ through the mapped ring header. Per-frame CultNet documents are neither needed
nor desirable.
```csharp
var initial = await subscriptions.SubscribeAsync(
"pilot-world",
recordKeys: new[] { entityLayoutRecord, bodyPublicationRecord },
schemaIds: new[] { entityLayoutSchema, CultMeshBodyPublicationSchemaVersions.Publication },
consumerRuntimeId: "eve-unity",
bodyIds: new[] { "eve:entity-soa:aetheria.daemon:pilot" },
supportedBodyTransports: new[]
{
CultMeshBodyTransportKind.SharedMemory.ToString(),
CultMeshBodyTransportKind.Network.ToString()
});
using var entities = await CultMesh.SubscribeLiveBodyAsync(
subscriptions,
bodyResolver,
new CultMeshLiveBodySubscription(
"pilot-entities",
"eve-unity",
"eve:entity-soa:aetheria.daemon:pilot"));
var bootstrap = initial.OfType<CultMeshBodyPublicationDocument>().Single();
var mapped = bootstrap.Representations.Single(CultMeshMappedFrameBodyCursor.CanOpen);
using var cursor = new CultMeshMappedFrameBodyCursor(mapped);
using var input = await subscriptions.SubscribeLiveValueAsync<EveInputCapabilityDocument>(
"pilot-input",
"eve:input:aetheria.daemon:pilot");
// Call from the consumer's normal update loop. The returned lease is read-only
// and must remain alive until the renderer has finished consuming that frame.
if (cursor.TryAcquireLatest(out var frame))
Present(frame);
// Open after each descriptor change. CultMesh chooses shared memory, shared-file
// mapping, or network from observed locality and the resolver's real adapters.
using var frame = entities.OpenCurrentReadOnly().Lease;
Present(frame, input.Current);
```
`CultMeshLiveBody` watches only the small latest-publication record and declares
the logical body demand. It never routes the body through the subscription
snapshot. `CultNetLiveValue<T>` does the corresponding job for exact scalar or
small structured state without creating a renderer-owned replica cache.
Snapshots remain bootstrap and recovery machinery. Reactive document changes
carry small control state; mapped or negotiated network bodies carry hot data.

6
src/GameCult.Mesh/docs/public-api.md
View file

@ -520,6 +520,12 @@ explicitly forwarded. Stale epochs fail loudly.
Consumers subscribe to typed document changes through `CultNetDatabase` watch
methods. Subscriptions receive domain changes rather than storage envelopes.
Exact remote values use `SubscribeLiveValueAsync<T>(...)`; they remain ephemeral
and own their unsubscribe lifetime. Hot body state uses
`CultMesh.SubscribeLiveBodyAsync(...)`: the reactive value is only the current
body descriptor, while `OpenCurrentReadOnly()` negotiates shared memory,
shared-file mapping, or network from observed locality and available adapters.
Body bytes never become subscription snapshot payload.
### Client Prediction

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

@ -899,7 +899,9 @@ one representation. A renderer owns none of those decisions.
**Inputs:** consumer runtime identity, exact logical body IDs, advertised
consumer transport capabilities, and server-observed peer locality. A
same-machine claim selects shared memory only when the transport peer is
actually loopback and the consumer explicitly supports shared memory.
actually loopback and the consumer explicitly supports shared memory. When
named shared memory is unavailable but the consumer can map files, the same
locality proof selects shared-file mapping before network.
**Outputs:** a per-body plan requiring shared memory, network, both, or no
publication. Producers use that plan before allocating, copying, hashing, or
@ -918,7 +920,15 @@ 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
`CultMesh.SubscribeLiveBodyAsync(...)` is the narrow public construction path.
Its `CultMeshLiveBodySubscription` names only the logical body, consumer, and
subscription lifetime. `CultMeshLiveBody` owns the ephemeral latest-publication
value and opens the current generation through the fastest valid representation
the resolver can actually import. Layout and other small state are independent
exact reactive values, so changing one variable does not redeliver an unrelated
view document or body.
`CultMesh.SubscribeHotBodyAsync(...)` remains the grouped compatibility 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

82
tests/GameCult.Mesh.Tests/CultMeshBodyPublicationTests.cs
View file

@ -1,7 +1,11 @@
using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Threading.Tasks;
using FluentAssertions;
using GameCult.Caching;
using GameCult.Networking;
using NUnit.Framework;
#nullable enable
@ -183,6 +187,39 @@ public sealed class CultMeshBodyPublicationTests
wrongGeneration.Should().Throw<InvalidOperationException>().WithMessage("*epoch*");
}
[Test]
public async Task LiveBody_SubscriptionKeepsBytesOffTheSnapshotAndOpensTheBestPlane()
{
var now = DateTimeOffset.UtcNow;
var bytes = new byte[] { 3, 1, 4, 1, 5, 9 };
var publication = Publication(bytes, now);
var cache = new CultCache();
var documents = CultMesh.CreateBodyPublicationDocumentRegistry(cache.Registry);
var transport = new BodyPublicationSchemaClient(publication, documents);
using var subscriptions = new CultNetDatabaseSubscriptionClient(transport, cache, documents);
using var body = await CultMesh.SubscribeLiveBodyAsync(
subscriptions,
Resolver(bytes, (_, _) => true),
new CultMeshLiveBodySubscription("pilot-body", "eve-unity", publication.BodyId));
transport.Subscribe.Should().NotBeNull();
transport.Subscribe!.RecordKeys.Should().Equal(
CultMeshBodyPublicationDocument.CreateLatestRecordKey(publication.BodyId).Value);
transport.Subscribe.SchemaIds.Should().Equal(CultMeshBodyPublicationSchemaVersions.Publication);
transport.Subscribe.BodyIds.Should().Equal(publication.BodyId);
transport.Subscribe.ConsumerRuntimeId.Should().Be("eve-unity");
cache.Get<CultMeshBodyPublicationDocument>(
CultMeshBodyPublicationDocument.CreateLatestRecordKey(publication.BodyId)).Should().BeNull(
"live descriptor control state must not become a renderer-owned replica");
using var opened = body.OpenCurrentReadOnly(now).Lease;
opened.TransportKind.Should().Be(CultMeshBodyTransportKind.SharedFileMapping);
var restored = new byte[bytes.Length];
opened.CopyTo(0, restored, 0, restored.Length).Should().Be(bytes.Length);
restored.Should().Equal(bytes);
}
private CultMeshBodyPublicationDocument Publication(byte[] bytes, DateTimeOffset now)
{
var local = new CultMeshMappedBodyPublisher(_root, TimeSpan.FromMinutes(1))
@ -248,4 +285,49 @@ public sealed class CultMeshBodyPublicationTests
CapabilityToken = "cultnet:body:aetheria:entities:7:42",
SemanticHash = local.SemanticHash
};
private sealed class BodyPublicationSchemaClient : ICultNetSchemaClient
{
private readonly CultMeshBodyPublicationDocument _publication;
private readonly CultNetDocumentRegistry _documents;
private readonly List<Action<CultNetSnapshotResponseRawMessage>> _snapshots = new();
public BodyPublicationSchemaClient(
CultMeshBodyPublicationDocument publication,
CultNetDocumentRegistry documents)
{
_publication = publication;
_documents = documents;
}
public bool Connected => true;
public CultNetDatabaseSubscribeMessage? Subscribe { get; private set; }
public void Connect(string host, int port) { }
public void SendCultNet<T>(T message) where T : ICultNetSchemaMessage
{
if (message is not CultNetDatabaseSubscribeMessage subscribe)
return;
Subscribe = subscribe;
var put = _documents.CreateRawDocumentPutMessage(
"body-publication",
new CultRecordHandle<CultMeshBodyPublicationDocument>(
CultMeshBodyPublicationDocument.CreateLatestRecordKey(_publication.BodyId)),
_publication);
var response = new CultNetSnapshotResponseRawMessage
{
MessageId = subscribe.MessageId,
Documents = new[] { put.Document }
};
foreach (var snapshot in _snapshots.ToArray()) snapshot(response);
}
public void OnCultNet<T>(Action<T> callback) where T : ICultNetSchemaMessage
{
if (typeof(T) == typeof(CultNetSnapshotResponseRawMessage))
_snapshots.Add(response => callback((T)(object)response));
}
public void Dispose() { }
}
}

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

@ -1,6 +1,7 @@
using System;
using System.IO;
using System.Linq;
using System.Threading.Tasks;
using FluentAssertions;
using NUnit.Framework;
using System.IO.MemoryMappedFiles;
@ -358,6 +359,48 @@ public sealed class CultMeshBodyTransportTests
plan.Consumers.Should().ContainSingle().Which.ConsumerRuntimeId.Should().Be("unknown");
}
[Test]
public void BodyDemand_UsesSharedFileMappingAsTheSameMachineFallback()
{
using var tracker = new CultMeshBodyDemandTracker();
tracker.Observe(new CultNetDatabaseSubscriptionDemand(
"unity-local",
"mapped-local",
new[] { "eve:entity-soa:aetheria.daemon:pilot" },
new[]
{
CultMeshBodyTransportKind.SharedFileMapping.ToString(),
CultMeshBodyTransportKind.Network.ToString()
},
sameMachine: true,
active: true));
var plan = tracker.Plan("eve:entity-soa:aetheria.daemon:pilot");
plan.RequiresSharedMemory.Should().BeFalse();
plan.RequiresSharedFileMapping.Should().BeTrue();
plan.RequiresNetwork.Should().BeFalse();
plan.Consumers.Should().ContainSingle().Which.Transport
.Should().Be(CultMeshBodyTransportKind.SharedFileMapping);
}
[Test]
public void LiveBodySubscriptionNamesOnlyTheReactiveDescriptorAndLogicalBodyDemand()
{
var subscription = new CultMeshLiveBodySubscription(
"pilot-entities",
"eve-unity",
"eve:body:pilot");
subscription.SubscriptionId.Should().Be("pilot-entities");
subscription.ConsumerRuntimeId.Should().Be("eve-unity");
subscription.BodyId.Should().Be("eve:body:pilot");
subscription.PublicationRecordKey.Should().Be(
CultMeshBodyPublicationDocument.CreateLatestRecordKey("eve:body:pilot").Value);
typeof(CultMesh).GetMethod(nameof(CultMesh.SubscribeLiveBodyAsync))!
.ReturnType.Should().Be(typeof(Task<CultMeshLiveBody>));
}
[Test]
public void HotBodySubscriptionKeepsViewPublicationAndDemandInOneExactContract()
{

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.27",
"version": "1.0.28",
"unity": "2021.3",
"description": "CultCache, CultNet, and CultMesh managed runtime assemblies for Unity.",
"license": "MIT",