From d6ac03071d3a221c485beb03d514fc39e849f290 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sun, 19 Jul 2026 22:15:50 +0100 Subject: [PATCH] Stream Aetheria realtime SoA over QUIC --- .../Aetheria.State.Daemon.csproj | 1 + Aetheria.State.Daemon/Program.cs | 130 +++++++++++++++--- .../AetheriaRuntimeDaemonSoaFramePublisher.cs | 44 +++++- docs/aetheria-perfect-machine-map.md | 23 ++++ .../Aetheria.TransportProbe.csproj | 14 ++ tools/Aetheria.TransportProbe/Program.cs | 40 ++++++ 6 files changed, 230 insertions(+), 22 deletions(-) create mode 100644 tools/Aetheria.TransportProbe/Aetheria.TransportProbe.csproj create mode 100644 tools/Aetheria.TransportProbe/Program.cs diff --git a/Aetheria.State.Daemon/Aetheria.State.Daemon.csproj b/Aetheria.State.Daemon/Aetheria.State.Daemon.csproj index 58693d1ac..0b7f6e8dc 100644 --- a/Aetheria.State.Daemon/Aetheria.State.Daemon.csproj +++ b/Aetheria.State.Daemon/Aetheria.State.Daemon.csproj @@ -11,6 +11,7 @@ + diff --git a/Aetheria.State.Daemon/Program.cs b/Aetheria.State.Daemon/Program.cs index 3c4b18fb3..9ae54b766 100644 --- a/Aetheria.State.Daemon/Program.cs +++ b/Aetheria.State.Daemon/Program.cs @@ -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 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 TickAsync( @@ -667,7 +678,7 @@ static async Task ImportRemoteCommittedFactsAsy return import.Tick; } -static AetheriaClientCultMeshHost StartClientCultMeshHost( +static async Task 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((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 PeerCultMeshEndpoints { get; init; } = Array.Empty(); @@ -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), diff --git a/Packages/org.gamecult.aetheria.state/Runtime/AetheriaRuntimeDaemonSoaFramePublisher.cs b/Packages/org.gamecult.aetheria.state/Runtime/AetheriaRuntimeDaemonSoaFramePublisher.cs index 226323b74..3db4aead0 100644 --- a/Packages/org.gamecult.aetheria.state/Runtime/AetheriaRuntimeDaemonSoaFramePublisher.cs +++ b/Packages/org.gamecult.aetheria.state/Runtime/AetheriaRuntimeDaemonSoaFramePublisher.cs @@ -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 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 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 PublishAsync(AetheriaRuntimeDaemonSoaFrame frame) + public Task 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.Empty; + return Task.FromResult(new AetheriaRuntimeDaemonSoaPublication(publication, view, realtimePayload)); + } + + private static ReadOnlyMemory 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(); diff --git a/docs/aetheria-perfect-machine-map.md b/docs/aetheria-perfect-machine-map.md index 61900cb7a..997eeb5b7 100644 --- a/docs/aetheria-perfect-machine-map.md +++ b/docs/aetheria-perfect-machine-map.md @@ -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. diff --git a/tools/Aetheria.TransportProbe/Aetheria.TransportProbe.csproj b/tools/Aetheria.TransportProbe/Aetheria.TransportProbe.csproj new file mode 100644 index 000000000..4cc13aa39 --- /dev/null +++ b/tools/Aetheria.TransportProbe/Aetheria.TransportProbe.csproj @@ -0,0 +1,14 @@ + + + + Exe + net10.0 + enable + enable + + + + + + diff --git a/tools/Aetheria.TransportProbe/Program.cs b/tools/Aetheria.TransportProbe/Program.cs new file mode 100644 index 000000000..1f6fcafc7 --- /dev/null +++ b/tools/Aetheria.TransportProbe/Program.cs @@ -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}.");