Stream Aetheria realtime SoA over QUIC

This commit is contained in:
Meta Vens 2026-07-19 22:15:50 +01:00
commit d6ac03071d
6 changed files with 230 additions and 22 deletions

View file

@ -11,6 +11,7 @@
<ItemGroup>
<ProjectReference Include="..\Aetheria.State\Aetheria.State.csproj" />
<ProjectReference Include="$(CultLibRoot)\src\GameCult.Mesh.Quic\GameCult.Mesh.Quic.csproj" />
<ProjectReference Include="$(YmirRoot)\src\Ymir.Core\Ymir.Core.csproj" />
</ItemGroup>

View file

@ -5,6 +5,7 @@ using GameCult.Aetheria.State.Verse;
using GameCult.Caching;
using GameCult.Eve.Surface;
using GameCult.Mesh;
using GameCult.Mesh.Quic;
using GameCult.Networking;
using MessagePack;
using System.Globalization;
@ -12,6 +13,8 @@ using System.Diagnostics;
using System.Net;
using System.Net.Sockets;
using System.Collections.Concurrent;
using System.Security.Cryptography;
using System.Security.Cryptography.X509Certificates;
var options = AetheriaDaemonHostOptions.Parse(args);
var startedAtUtc = DateTimeOffset.UtcNow.ToString("O");
@ -52,10 +55,12 @@ var latestFrame = await node.MutableDocument<AetheriaRuntimeDaemonFrameDocument>
TraceStartup("latest-frame");
var unityBundles = BuildUnityBundleArtifactSet(options);
TraceStartup("provider-asset-bundles");
using var cultMeshClientHost = StartClientCultMeshHost(node, options, unityBundles, () => latestFrame);
await using var cultMeshClientHost = await StartClientCultMeshHostAsync(
node, options, unityBundles, () => latestFrame).ConfigureAwait(false);
TraceStartup("client-host");
Console.WriteLine($"Aetheria client CultMesh endpoint: {cultMeshClientHost.ControlEndpoint}");
Console.WriteLine($"Aetheria client CDN endpoint: {cultMeshClientHost.ContentEndpoint}");
Console.WriteLine($"Aetheria client realtime endpoint: {cultMeshClientHost.RealtimeEndpoint}");
Console.WriteLine($"Aetheria daemon transport-ready in {startup.Elapsed.TotalMilliseconds:0.###}ms.");
using var clientSubscriptions = new CultNetDatabaseSubscriptionServer(cultMeshClientHost.Control, node.Database);
var managedViewportDemand = new AetheriaManagedViewportDemandState();
@ -120,7 +125,8 @@ await PublishPreparedDocumentsAsync(
reactiveSurfaceState,
publishTopology: true).ConfigureAwait(false);
TraceStartup("first-publication");
await PublishHotEntityStateAsync(node, soaPublisher, hotState, firstTick.Frame, ingressState.Catalog).ConfigureAwait(false);
await PublishHotEntityStateAsync(
node, soaPublisher, hotState, firstTick.Frame, ingressState.Catalog, cultMeshClientHost).ConfigureAwait(false);
TraceStartup("hot-entity-state");
nextApiPublicationUtc = DateTimeOffset.UtcNow.Add(options.ApiPublicationInterval);
Console.WriteLine($"Aetheria Verse daemon published frame {firstTick.Frame.FrameId}.");
@ -158,7 +164,8 @@ while (!stopped.Task.IsCompleted)
var tick = await TickAsync(node, options, unityBundles, worldPhysics, latestFrame, ingressState, buildPublications: false).ConfigureAwait(false);
ThrowIfClientHostFaulted(cultMeshClientHost);
latestFrame = tick.Frame;
await PublishHotEntityStateAsync(node, soaPublisher, hotState, tick.Frame, ingressState.Catalog).ConfigureAwait(false);
await PublishHotEntityStateAsync(
node, soaPublisher, hotState, tick.Frame, ingressState.Catalog, cultMeshClientHost).ConfigureAwait(false);
nextTickUtc += options.TickInterval;
if (nextTickUtc < DateTimeOffset.UtcNow - options.TickInterval)
nextTickUtc = DateTimeOffset.UtcNow;
@ -188,6 +195,7 @@ while (!stopped.Task.IsCompleted)
Console.WriteLine(
$"Aetheria Verse daemon published frame {tick.Frame.FrameId} at {tick.Frame.SimulationTimeSeconds:0.00}s; " +
$"control peers={cultMeshClientHost.Control.PeerCount} " +
$"quic peers={cultMeshClientHost.Realtime.ConnectionCount} " +
$"cdn chunks={cultMeshClientHost.Content.ChunkRequestsServed}.");
}
}
@ -198,11 +206,14 @@ Console.WriteLine("Aetheria Verse daemon stopping.");
static void ThrowIfClientHostFaulted(AetheriaClientCultMeshHost host)
{
if (!host.Control.BackgroundFailure.IsCompleted)
return;
throw new InvalidOperationException(
"Aetheria client CultMesh TCP control host faulted.",
host.Control.BackgroundFailure.GetAwaiter().GetResult());
if (host.Control.BackgroundFailure.IsCompleted)
throw new InvalidOperationException(
"Aetheria client CultMesh TCP control host faulted.",
host.Control.BackgroundFailure.GetAwaiter().GetResult());
if (host.Realtime.BackgroundFailure.IsCompleted)
throw new InvalidOperationException(
"Aetheria client CultMesh QUIC realtime host faulted.",
host.Realtime.BackgroundFailure.GetAwaiter().GetResult());
}
static async Task<AetheriaRuntimeDaemonTickResult> TickAsync(
@ -667,7 +678,7 @@ static async Task<AetheriaRuntimeDaemonTickResult> ImportRemoteCommittedFactsAsy
return import.Tick;
}
static AetheriaClientCultMeshHost StartClientCultMeshHost(
static async Task<AetheriaClientCultMeshHost> StartClientCultMeshHostAsync(
AetheriaStateNode node,
AetheriaDaemonHostOptions options,
AetheriaUnityBundleArtifactSet unityBundles,
@ -683,9 +694,33 @@ static AetheriaClientCultMeshHost StartClientCultMeshHost(
var content = new CultMeshTcpContentServer(new TcpListener(
ParseBindAddress(options.ClientCultMeshHost),
options.ClientCultMeshContentPort), contentCache);
var realtimeCertificate = CreateRealtimeCertificate(options.ClientCultMeshAdvertiseHost);
CultMeshQuicRealtimeServer realtime;
try
{
realtime = await CultMeshQuicRealtimeServer.ListenAsync(new CultMeshQuicRealtimeServerOptions
{
ListenEndPoint = new IPEndPoint(
ParseBindAddress(options.ClientCultMeshHost),
options.ClientCultMeshQuicPort),
ServerCertificate = realtimeCertificate
}).ConfigureAwait(false);
}
catch
{
realtimeCertificate.Dispose();
content.Dispose();
server.Dispose();
contentCache.Dispose();
throw;
}
var advertisedEndpoint = $"cultnet+tcp://{options.ClientCultMeshAdvertiseHost}:{server.LocalEndPoint.Port}";
var advertisedContentEndpoint =
$"{CultMeshTcpContentTransportConnector.Scheme}://{options.ClientCultMeshAdvertiseHost}:{content.LocalEndPoint.Port}";
var certificatePin = Convert.ToHexString(SHA256.HashData(realtimeCertificate.RawData));
var advertisedRealtimeEndpoint =
$"{CultMeshQuicRealtimeTransportConnector.Scheme}://{options.ClientCultMeshAdvertiseHost}:{realtime.LocalEndPoint.Port}" +
$"?cert-sha256={certificatePin}";
server.PeerFailed += (endpoint, error) => Console.Error.WriteLine(
$"Aetheria client CultMesh peer {endpoint} failed: {error.GetType().Name}: {error.Message}");
server.OnCultNet<CultMeshVerseCatalogRequestMessage>((request, peer) =>
@ -697,7 +732,7 @@ static AetheriaClientCultMeshHost StartClientCultMeshHost(
new CultMeshVerseCompatibility(
"cultmesh.v0",
CultMeshVerseDescriptor.ComputeRulesHash("aetheria", "runtime-world.v1")),
discoveryEndpoints: new[] { advertisedEndpoint, advertisedContentEndpoint },
discoveryEndpoints: new[] { advertisedEndpoint, advertisedContentEndpoint, advertisedRealtimeEndpoint },
authorityRuntimeIds: new[] { options.DaemonId },
description: "Aetheria provider Verse");
peer.SendCultNet(new CultMeshVerseCatalogResponseMessage
@ -1136,9 +1171,36 @@ static AetheriaClientCultMeshHost StartClientCultMeshHost(
return new AetheriaClientCultMeshHost(
server,
content,
realtime,
realtimeCertificate,
contentCache,
advertisedEndpoint,
advertisedContentEndpoint);
advertisedContentEndpoint,
advertisedRealtimeEndpoint);
}
static X509Certificate2 CreateRealtimeCertificate(string advertisedHost)
{
using var key = RSA.Create(2048);
var request = new CertificateRequest(
$"CN={advertisedHost}",
key,
HashAlgorithmName.SHA256,
RSASignaturePadding.Pkcs1);
request.CertificateExtensions.Add(new X509BasicConstraintsExtension(false, false, 0, false));
request.CertificateExtensions.Add(new X509KeyUsageExtension(X509KeyUsageFlags.DigitalSignature, false));
request.CertificateExtensions.Add(new X509EnhancedKeyUsageExtension(
new OidCollection { new("1.3.6.1.5.5.7.3.1") },
false));
var names = new SubjectAlternativeNameBuilder();
if (IPAddress.TryParse(advertisedHost, out var address)) names.AddIpAddress(address);
else names.AddDnsName(advertisedHost);
request.CertificateExtensions.Add(names.Build());
request.CertificateExtensions.Add(new X509SubjectKeyIdentifierExtension(request.PublicKey, false));
using var generated = request.CreateSelfSigned(
DateTimeOffset.UtcNow.AddMinutes(-1),
DateTimeOffset.UtcNow.AddDays(7));
return X509CertificateLoader.LoadPkcs12(generated.Export(X509ContentType.Pfx), null);
}
static void InjectCultMeshCdnManifestSnapshots(
@ -1612,13 +1674,20 @@ static async Task PublishHotEntityStateAsync(
AetheriaRuntimeDaemonSoaFramePublisher publisher,
AetheriaHotEntityPublicationState state,
AetheriaRuntimeDaemonFrameDocument frame,
AetheriaRuntimeCatalogSnapshot? catalog)
AetheriaRuntimeCatalogSnapshot? catalog,
AetheriaClientCultMeshHost clientHost)
{
using var hotFrame = publisher.BuildCurrentZoneEntities(frame, catalog);
var hasRealtimeConsumers = clientHost.Realtime.ConnectionCount > 0;
using var hotFrame = publisher.BuildCurrentZoneEntities(
frame,
catalog,
realtimeDemand: hasRealtimeConsumers);
if (hotFrame == null)
return;
var publication = await publisher.PublishAsync(hotFrame).ConfigureAwait(false);
var publication = await publisher.PublishAsync(
hotFrame,
includeRealtimePayload: hasRealtimeConsumers).ConfigureAwait(false);
var layoutChanged = !state.Matches(publication.View);
if (layoutChanged)
{
@ -1638,6 +1707,19 @@ static async Task PublishHotEntityStateAsync(
.ConfigureAwait(false);
state.Set(publication.View);
}
if (!publication.RealtimePayload.IsEmpty)
{
await clientHost.Realtime.BroadcastAsync(new CultMeshRealtimeFrame
{
ChannelId = "aetheria.entities",
SchemaId = AetheriaRuntimeDaemonSoaFramePublisher.BodySchemaId,
BodyId = AetheriaRuntimeDaemonSoaFramePublisher.BodyId,
ProducerEpoch = publication.Body.ProducerEpoch,
Sequence = publication.Body.Sequence,
Delivery = CultMeshRealtimeDelivery.LatestOnly,
Payload = publication.RealtimePayload
}).ConfigureAwait(false);
}
}
static async Task PublishDaemonApiDocumentsAsync(
@ -3517,30 +3599,41 @@ internal sealed class AetheriaUnityBundleArtifact
public CultMeshCdnArtifact Artifact { get; }
}
internal sealed class AetheriaClientCultMeshHost : IDisposable
internal sealed class AetheriaClientCultMeshHost : IAsyncDisposable
{
public AetheriaClientCultMeshHost(
TcpFramedCultNetSchemaServer control,
CultMeshTcpContentServer content,
CultMeshQuicRealtimeServer realtime,
X509Certificate2 realtimeCertificate,
CultCache contentCache,
string controlEndpoint,
string contentEndpoint)
string contentEndpoint,
string realtimeEndpoint)
{
Control = control ?? throw new ArgumentNullException(nameof(control));
Content = content ?? throw new ArgumentNullException(nameof(content));
Realtime = realtime ?? throw new ArgumentNullException(nameof(realtime));
RealtimeCertificate = realtimeCertificate ?? throw new ArgumentNullException(nameof(realtimeCertificate));
ContentCache = contentCache ?? throw new ArgumentNullException(nameof(contentCache));
ControlEndpoint = controlEndpoint ?? throw new ArgumentNullException(nameof(controlEndpoint));
ContentEndpoint = contentEndpoint ?? throw new ArgumentNullException(nameof(contentEndpoint));
RealtimeEndpoint = realtimeEndpoint ?? throw new ArgumentNullException(nameof(realtimeEndpoint));
}
public TcpFramedCultNetSchemaServer Control { get; }
public CultMeshTcpContentServer Content { get; }
public CultMeshQuicRealtimeServer Realtime { get; }
private X509Certificate2 RealtimeCertificate { get; }
private CultCache ContentCache { get; }
public string ControlEndpoint { get; }
public string ContentEndpoint { get; }
public string RealtimeEndpoint { get; }
public void Dispose()
public async ValueTask DisposeAsync()
{
await Realtime.DisposeAsync().ConfigureAwait(false);
RealtimeCertificate.Dispose();
Content.Dispose();
ContentCache.Dispose();
Control.Dispose();
@ -3558,6 +3651,7 @@ internal sealed class AetheriaDaemonHostOptions
public string ClientCultMeshAdvertiseHost { get; init; } = "127.0.0.1";
public int ClientCultMeshPort { get; init; } = 3076;
public int ClientCultMeshContentPort { get; init; }
public int ClientCultMeshQuicPort { get; init; }
public string AetheriaResourcesRoot { get; init; } = "";
public string AssetBundleRoot { get; init; } = "";
public IReadOnlyList<string> PeerCultMeshEndpoints { get; init; } = Array.Empty<string>();
@ -3591,6 +3685,7 @@ internal sealed class AetheriaDaemonHostOptions
var clientCultMeshAdvertiseHost = ReadOption(args, "--client-cultmesh-advertise-host");
var clientCultMeshPort = ReadNonNegativeInt(args, "--client-cultmesh-port") ?? 3076;
var clientCultMeshContentPort = ReadNonNegativeInt(args, "--client-cultmesh-content-port") ?? 0;
var clientCultMeshQuicPort = ReadNonNegativeInt(args, "--client-cultmesh-quic-port") ?? 0;
var aetheriaResourcesRoot = ReadOption(args, "--aetheria-resources-root");
var assetBundleRoot = ReadOption(args, "--asset-bundle-root");
RejectRemovedOption(args, "--rts-cultmesh-port", "--client-cultmesh-port");
@ -3622,6 +3717,7 @@ internal sealed class AetheriaDaemonHostOptions
: clientCultMeshAdvertiseHost,
ClientCultMeshPort = clientCultMeshPort,
ClientCultMeshContentPort = clientCultMeshContentPort,
ClientCultMeshQuicPort = clientCultMeshQuicPort,
AetheriaResourcesRoot = string.IsNullOrWhiteSpace(aetheriaResourcesRoot)
? Path.GetFullPath(Path.Combine(root, "Assets", "Resources"))
: Path.GetFullPath(aetheriaResourcesRoot),

View file

@ -1,6 +1,7 @@
using System;
using System.Buffers.Binary;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Threading.Tasks;
using GameCult.Eve.Surface;
@ -46,14 +47,17 @@ namespace GameCult.Aetheria.State.Verse
{
public AetheriaRuntimeDaemonSoaPublication(
CultMeshBodyPublicationDocument body,
EveEntitySoaViewDocument view)
EveEntitySoaViewDocument view,
ReadOnlyMemory<byte> realtimePayload)
{
Body = body ?? throw new ArgumentNullException(nameof(body));
View = view ?? throw new ArgumentNullException(nameof(view));
RealtimePayload = realtimePayload;
}
public CultMeshBodyPublicationDocument Body { get; }
public EveEntitySoaViewDocument View { get; }
public ReadOnlyMemory<byte> RealtimePayload { get; }
}
public sealed class AetheriaRuntimeDaemonSoaFramePublisher : IDisposable
@ -92,12 +96,15 @@ namespace GameCult.Aetheria.State.Verse
public AetheriaRuntimeDaemonSoaFrame? BuildCurrentZoneEntities(
AetheriaRuntimeDaemonFrameDocument frame,
AetheriaRuntimeCatalogSnapshot? catalog = null)
AetheriaRuntimeCatalogSnapshot? catalog = null,
bool realtimeDemand = false)
{
if (frame == null)
throw new ArgumentNullException(nameof(frame));
var demand = _demand?.Plan(BodyId);
if (demand != null && (!demand.HasConsumers || !demand.RequiresSharedMemory))
if (demand != null &&
(!demand.HasConsumers || !demand.RequiresSharedMemory) &&
!realtimeDemand)
return null;
var run = frame.Run ?? new AetheriaRuntimeRunCheckpointCommit();
@ -281,7 +288,9 @@ namespace GameCult.Aetheria.State.Verse
}
}
public Task<AetheriaRuntimeDaemonSoaPublication> PublishAsync(AetheriaRuntimeDaemonSoaFrame frame)
public Task<AetheriaRuntimeDaemonSoaPublication> PublishAsync(
AetheriaRuntimeDaemonSoaFrame frame,
bool includeRealtimePayload = false)
{
if (frame == null) throw new ArgumentNullException(nameof(frame));
var now = DateTimeOffset.UtcNow;
@ -318,7 +327,32 @@ namespace GameCult.Aetheria.State.Verse
if (view.Buffers.Length != 1 || !string.Equals(view.Buffers[0].BufferId, publication.BodyId, StringComparison.Ordinal) ||
view.Columns.Any(column => !string.Equals(column.BufferId, publication.BodyId, StringComparison.Ordinal)))
throw new InvalidOperationException("Aetheria Eve SoA layout does not reference its published logical body identity.");
return Task.FromResult(new AetheriaRuntimeDaemonSoaPublication(publication, view));
var realtimePayload = includeRealtimePayload
? CopyRealtimePayload(local, now)
: ReadOnlyMemory<byte>.Empty;
return Task.FromResult(new AetheriaRuntimeDaemonSoaPublication(publication, view, realtimePayload));
}
private static ReadOnlyMemory<byte> CopyRealtimePayload(CultMeshBodyDescriptor descriptor, DateTimeOffset now)
{
var bytes = new byte[checked((int)descriptor.ByteSize)];
using var lease = new CultMeshSharedMemoryBodyAdapter().OpenReadOnly(
descriptor,
new CultMeshBodyValidationRequest
{
BodyId = descriptor.BodyId,
SchemaId = descriptor.SchemaId,
LayoutVersion = descriptor.LayoutVersion,
ProducerEpoch = descriptor.ProducerEpoch,
Sequence = descriptor.Sequence,
Capacity = descriptor.Capacity,
AccessMode = CultMeshBodyAccessMode.ReadOnly,
NowUtc = now
});
var copied = lease.CopyTo(0, bytes, 0, bytes.Length);
if (copied != bytes.Length)
throw new EndOfStreamException($"CultMesh copied {copied} of {bytes.Length} realtime SoA bytes.");
return bytes;
}
public void Dispose() => _localPublisher.Dispose();

View file

@ -3576,3 +3576,26 @@ Ymir restart ownership is deliberately private and asymmetric:
the next tick cannot resurrect the pickup or duplicate cargo or feedback. Ymir unit
tests prove incremental suffix capture, exact chunk coverage, checksums, and
replay verifier convergence.
Remote realtime transport has one narrow owner:
- Owner: the Aetheria client CultMesh host owns the provider QUIC listener;
CultMesh session selection owns the matching client connector and route.
- Inputs: the exact committed portable Eve entity SoA body, its schema id,
producer epoch, and sequence. Trusted discovery supplies the provider's
short-lived TLS certificate SHA-256 pin with the QUIC route.
- Outputs: latest-only `aetheria.entities` frames over QUIC when at least one
remote realtime consumer is connected.
- Derived state: connection count and the temporary network payload copy are
delivery state only. They do not own simulation or world truth.
- Forbidden writers: QUIC clients, TCP snapshot consumers, Unity lowerers, and
legacy RUDP cannot write or repair authoritative entity state.
- Shared paths: the same committed SoA publication feeds mapped memory for a
same-machine client and QUIC for a remote client. No QUIC payload is copied
when no remote consumer exists.
- Cut line: RUDP is not selected or advertised; TCP retains schemas, commands,
receipts, manifests, and immutable content rather than carrying hot SoA
snapshots.
- Verification layer: `Aetheria.TransportProbe` discovers the daemon through
its TCP catalog, authenticates the advertised QUIC route, and validates a
nonempty latest-only frame with the exact Aetheria body and schema identity.

View file

@ -0,0 +1,14 @@
<Project Sdk="Microsoft.NET.Sdk">
<Import Project="..\..\Aetheria.State\Aetheria.State.Dependencies.props" />
<PropertyGroup>
<OutputType>Exe</OutputType>
<TargetFramework>net10.0</TargetFramework>
<Nullable>enable</Nullable>
<ImplicitUsings>enable</ImplicitUsings>
</PropertyGroup>
<ItemGroup>
<ProjectReference Include="..\..\Aetheria.State\Aetheria.State.csproj"
AdditionalProperties="CultLibRoot=$(CultLibRoot)" />
<ProjectReference Include="$(CultLibRoot)\src\GameCult.Mesh.Quic\GameCult.Mesh.Quic.csproj" />
</ItemGroup>
</Project>

View file

@ -0,0 +1,40 @@
using GameCult.Aetheria.State.Verse;
using GameCult.Mesh;
using GameCult.Mesh.Quic;
using GameCult.Networking;
var controlEndpoint = args.ElementAtOrDefault(0) ?? "cultnet+tcp://127.0.0.1:3076";
var verseId = args.ElementAtOrDefault(1) ?? "aetheria.local";
using var deadline = new CancellationTokenSource(TimeSpan.FromSeconds(15));
var discovery = new CultMeshVerseDiscoveryClient();
var catalog = await discovery.FetchAsync(
controlEndpoint,
new CultMeshVerseCatalogRequestMessage
{
VerseIds = new[] { verseId },
TransportVersion = "cultmesh.v0"
}).ConfigureAwait(false);
var advertised = catalog.Verses
.SingleOrDefault(candidate => string.Equals(candidate.VerseId, verseId, StringComparison.Ordinal))
?? throw new InvalidDataException($"Aetheria Verse '{verseId}' was not advertised by {controlEndpoint}.");
Console.WriteLine($"Aetheria advertised routes: {string.Join(", ", advertised.DiscoveryEndpoints)}");
using var client = new CultMeshClient(new CultMeshClientOptions
{
RendezvousEndpoints = new[] { controlEndpoint },
RealtimeConnectors = new ICultMeshRealtimeTransportConnector[]
{
new CultMeshQuicRealtimeTransportConnector()
}
});
using var session = await client.ConnectRealtimeAsync(verseId, deadline.Token).ConfigureAwait(false);
var frame = await session.ReceiveAsync(deadline.Token).ConfigureAwait(false);
if (!string.Equals(frame.BodyId, AetheriaRuntimeDaemonSoaFramePublisher.BodyId, StringComparison.Ordinal) ||
!string.Equals(frame.SchemaId, AetheriaRuntimeDaemonSoaFramePublisher.BodySchemaId, StringComparison.Ordinal) ||
frame.Payload.IsEmpty)
throw new InvalidDataException(
$"Aetheria QUIC frame mismatch: body={frame.BodyId} schema={frame.SchemaId} bytes={frame.Payload.Length}.");
Console.WriteLine(
$"Aetheria realtime transport={session.TransportId} delivery={frame.Delivery} " +
$"epoch={frame.ProducerEpoch} sequence={frame.Sequence} bytes={frame.Payload.Length}.");