diff --git a/packages/org.gamecult.eve.unity-scene/Runtime/CultMesh/EveUnityCultMeshLiveProviderTransport.cs b/packages/org.gamecult.eve.unity-scene/Runtime/CultMesh/EveUnityCultMeshLiveProviderTransport.cs index 5cc8a79..80a4539 100644 --- a/packages/org.gamecult.eve.unity-scene/Runtime/CultMesh/EveUnityCultMeshLiveProviderTransport.cs +++ b/packages/org.gamecult.eve.unity-scene/Runtime/CultMesh/EveUnityCultMeshLiveProviderTransport.cs @@ -3,6 +3,7 @@ using System.Collections.Generic; using System.IO; using System.Linq; using System.Collections.Concurrent; +using System.Threading; using System.Threading.Tasks; using GameCult.Caching; using GameCult.Caching.MessagePack; @@ -10,6 +11,7 @@ using GameCult.Eve.PluginFields; using GameCult.Eve.Surface; using GameCult.Eve.UnityScene.Fields; using GameCult.Mesh; +using GameCult.Mesh.Quic.Native; using GameCult.Networking; using UnityEngine; using Stopwatch = System.Diagnostics.Stopwatch; @@ -63,6 +65,11 @@ namespace GameCult.Eve.UnityScene private readonly List _assetBodyLeases = new List(); private readonly Dictionary _renderChannelLayers = new Dictionary(StringComparer.Ordinal); private readonly ConcurrentQueue _liveDocuments = new ConcurrentQueue(); + private readonly object _realtimeFrameGate = new object(); + private readonly Dictionary _latestRealtimeFrames = + new Dictionary(StringComparer.Ordinal); + private readonly HashSet _queuedRealtimeFrameBodies = + new HashSet(StringComparer.Ordinal); private readonly Dictionary _embeddedSurfaces = new Dictionary(StringComparer.Ordinal); private CultMeshNode? _node; @@ -84,6 +91,9 @@ namespace GameCult.Eve.UnityScene private CultMeshBodyPublicationDocument? _latestEntityPublication; private CultMeshMappedFrameBodyCursor? _mappedEntityFrameCursor; private CultMeshBodyDescriptor? _mappedEntityFrameContract; + private CancellationTokenSource? _realtimeLifetime; + private Task? _realtimePump; + private CultMeshRealtimeSession? _realtimeSession; private long _lastQueuedEntityViewEpoch = -1; private long _lastQueuedEntityViewSequence = -1; private long _lastPresentedEntityViewEpoch = -1; @@ -221,6 +231,8 @@ namespace GameCult.Eve.UnityScene { if (document is PendingEntityGeneration entityGeneration) PublishEntityView(entityGeneration.View, entityGeneration.Publication); + else if (document is PendingRealtimeEntityFrame realtimeFrame) + PublishPendingRealtimeEntityFrame(realtimeFrame.BodyId); else if (document is EveFieldsSplatsDocument fields) FieldsSplatsAvailable?.Invoke(fields); else if (document is PendingSurfaceDocument surface) @@ -398,6 +410,17 @@ namespace GameCult.Eve.UnityScene _mappedEntityFrameCursor?.Dispose(); _mappedEntityFrameCursor = null; _mappedEntityFrameContract = null; + _realtimeLifetime?.Cancel(); + _realtimeSession?.Dispose(); + _realtimeSession = null; + if (_realtimePump != null) + { + try { _realtimePump.GetAwaiter().GetResult(); } + catch (OperationCanceledException) { } + } + _realtimePump = null; + _realtimeLifetime?.Dispose(); + _realtimeLifetime = null; _entityBody?.Dispose(); _entityBody = null; _inputCapability?.Dispose(); @@ -502,7 +525,11 @@ namespace GameCult.Eve.UnityScene _networkRegistry); _meshClient = new CultMeshClient(new CultMeshClientOptions { - RendezvousEndpoints = new[] { _endpoint } + RendezvousEndpoints = new[] { _endpoint }, + RealtimeConnectors = new ICultMeshRealtimeTransportConnector[] + { + new CultMeshNativeQuicRealtimeTransportConnector() + } }); } @@ -595,35 +622,7 @@ namespace GameCult.Eve.UnityScene EnsureMappedEntityFrameCursor(publication); if (_mappedEntityFrameCursor != null) return; - _bodyResolver ??= CreateBodyResolver(); - var lease = _bodyResolver.ResolveReadOnly(publication, new CultMeshBodyValidationRequest - { - BodyId = handle.BodyId, - SchemaId = document.BodySchemaId, - LayoutVersion = document.LayoutVersion, - ProducerEpoch = document.ProducerEpoch, - Sequence = document.Sequence, - Capacity = document.Capacity, - AccessMode = CultMeshBodyAccessMode.ReadOnly, - NowUtc = now - }); - var handler = EntityViewAvailable; - if (handler == null) - { - lease.Dispose(); - return; - } - try - { - handler(document, lease); - _lastPresentedEntityViewEpoch = document.ProducerEpoch; - _lastPresentedEntityViewSequence = document.Sequence; - } - catch - { - lease.Dispose(); - throw; - } + EnsureRealtimeEntityState(); } private void EnsureMappedEntityFrameCursor(CultMeshBodyPublicationDocument publication) @@ -651,10 +650,196 @@ namespace GameCult.Eve.UnityScene return; _mappedEntityFrameCursor?.Dispose(); - _mappedEntityFrameCursor = new CultMeshMappedFrameBodyCursor(descriptor); - _mappedEntityFrameContract = descriptor; - TraceHotState( - $"mapped frame cursor body={descriptor.BodyId} epoch={descriptor.ProducerEpoch} sequence={descriptor.Sequence}"); + try + { + _mappedEntityFrameCursor = new CultMeshMappedFrameBodyCursor(descriptor); + _mappedEntityFrameContract = descriptor; + TraceHotState( + $"mapped frame cursor body={descriptor.BodyId} epoch={descriptor.ProducerEpoch} sequence={descriptor.Sequence}"); + } + catch (Exception error) when ( + error is IOException || + error is UnauthorizedAccessException || + error is InvalidOperationException) + { + _mappedEntityFrameCursor = null; + _mappedEntityFrameContract = null; + TraceHotState($"mapped frame unavailable; selecting remote realtime plane: {error.Message}"); + } + } + + private void EnsureRealtimeEntityState() + { + if (_realtimePump != null) + return; + if (_advertisement == null || string.IsNullOrWhiteSpace(_advertisement.VerseId)) + throw new InvalidOperationException( + $"Eve provider '{_providerId}' does not advertise the Verse identity required for realtime state."); + _realtimeLifetime = new CancellationTokenSource(); + _realtimePump = Task.Run(() => PumpRealtimeEntityStateAsync( + _advertisement.VerseId, + _realtimeLifetime.Token)); + } + + private async Task PumpRealtimeEntityStateAsync(string verseId, CancellationToken cancellationToken) + { + try + { + _realtimeSession = await _meshClient! + .ConnectRealtimeAsync(verseId, cancellationToken) + .ConfigureAwait(false); + TraceHotState($"realtime state connected transport={_realtimeSession.TransportId} verse={verseId}"); + while (!cancellationToken.IsCancellationRequested) + { + var frame = await _realtimeSession.ReceiveAsync(cancellationToken).ConfigureAwait(false); + EnqueueRealtimeEntityFrame(frame); + } + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + } + catch (ObjectDisposedException) when (_disposed) + { + } + catch (Exception error) + { + _liveDocuments.Enqueue(error); + } + } + + private void EnqueueRealtimeEntityFrame(CultMeshRealtimeFrame frame) + { + if (frame == null) + throw new ArgumentNullException(nameof(frame)); + lock (_realtimeFrameGate) + { + if (_latestRealtimeFrames.TryGetValue(frame.BodyId, out var current) && + (frame.ProducerEpoch < current.ProducerEpoch || + (frame.ProducerEpoch == current.ProducerEpoch && frame.Sequence <= current.Sequence))) + return; + _latestRealtimeFrames[frame.BodyId] = frame; + if (_queuedRealtimeFrameBodies.Add(frame.BodyId)) + _liveDocuments.Enqueue(new PendingRealtimeEntityFrame(frame.BodyId)); + } + } + + private void PublishPendingRealtimeEntityFrame(string bodyId) + { + CultMeshRealtimeFrame? frame; + lock (_realtimeFrameGate) + { + _queuedRealtimeFrameBodies.Remove(bodyId); + if (!_latestRealtimeFrames.TryGetValue(bodyId, out frame)) + return; + _latestRealtimeFrames.Remove(bodyId); + } + PublishRealtimeEntityFrame(frame); + } + + private void PublishRealtimeEntityFrame(CultMeshRealtimeFrame frame) + { + var layout = _latestEntityLayout + ?? throw new InvalidOperationException("Realtime entity state arrived before its Eve SoA layout."); + var publication = _latestEntityPublication + ?? throw new InvalidOperationException("Realtime entity state arrived before its body publication."); + if (!string.Equals(frame.BodyId, publication.BodyId, StringComparison.Ordinal) || + !string.Equals(frame.SchemaId, layout.BodySchemaId, StringComparison.Ordinal)) + throw new UnauthorizedAccessException("Realtime entity state does not match the advertised Eve body contract."); + if (frame.Delivery != CultMeshRealtimeDelivery.LatestOnly) + throw new InvalidDataException("Playable entity state requires latest-only QUIC delivery."); + if (frame.ProducerEpoch < _lastPresentedEntityViewEpoch || + (frame.ProducerEpoch == _lastPresentedEntityViewEpoch && + frame.Sequence <= _lastPresentedEntityViewSequence)) + return; + var representation = (publication.Representations ?? Array.Empty()) + .FirstOrDefault(value => + string.Equals(value.BodyId, frame.BodyId, StringComparison.Ordinal) && + string.Equals(value.SchemaId, frame.SchemaId, StringComparison.Ordinal)) + ?? throw new InvalidDataException("Realtime entity state has no advertised body representation."); + if (frame.Payload.Length != representation.ByteSize) + throw new InvalidDataException( + $"Realtime entity body size {frame.Payload.Length} does not match advertised size {representation.ByteSize}."); + + var generation = MaterializeEntityGeneration(layout, frame.ProducerEpoch, frame.Sequence); + var lease = new RealtimeEntityBodyLease( + new CultMeshBodyDescriptor + { + BodyId = frame.BodyId, + SchemaId = frame.SchemaId, + LayoutVersion = layout.LayoutVersion, + ByteSize = frame.Payload.Length, + Capacity = layout.Capacity, + ProducerEpoch = frame.ProducerEpoch, + Sequence = frame.Sequence, + AccessMode = CultMeshBodyAccessMode.ReadOnly, + Synchronization = CultMeshBodySynchronization.EpochSequence, + LeaseExpiresAtUnixMs = DateTimeOffset.UtcNow.AddSeconds(2).ToUnixTimeMilliseconds(), + TransportKind = CultMeshBodyTransportKind.Network, + CapabilityToken = "authenticated-quic", + SemanticHash = "" + }, + frame.Payload.ToArray()); + var handler = EntityViewAvailable; + if (handler == null) + { + lease.Dispose(); + return; + } + try + { + handler(generation, lease); + _lastPresentedEntityViewEpoch = frame.ProducerEpoch; + _lastPresentedEntityViewSequence = frame.Sequence; + } + catch + { + lease.Dispose(); + throw; + } + } + + private sealed class PendingRealtimeEntityFrame + { + public PendingRealtimeEntityFrame(string bodyId) => + BodyId = string.IsNullOrWhiteSpace(bodyId) + ? throw new ArgumentException("Body id must be non-empty.", nameof(bodyId)) + : bodyId; + + public string BodyId { get; } + } + + private sealed class RealtimeEntityBodyLease : ICultMeshBodyReadLease + { + private byte[] _bytes; + + public RealtimeEntityBodyLease(CultMeshBodyDescriptor descriptor, byte[] bytes) + { + Descriptor = descriptor; + _bytes = bytes; + } + + public CultMeshBodyDescriptor Descriptor { get; } + public CultMeshBodyTransportKind TransportKind => CultMeshBodyTransportKind.Network; + public byte ReadByte(long offset) { Validate(offset, 1); return _bytes[(int)offset]; } + public int ReadInt32(long offset) { Validate(offset, 4); return BitConverter.ToInt32(_bytes, (int)offset); } + public long ReadInt64(long offset) { Validate(offset, 8); return BitConverter.ToInt64(_bytes, (int)offset); } + public float ReadSingle(long offset) { Validate(offset, 4); return BitConverter.ToSingle(_bytes, (int)offset); } + public double ReadDouble(long offset) { Validate(offset, 8); return BitConverter.ToDouble(_bytes, (int)offset); } + + public int CopyTo(long offset, byte[] destination, int destinationOffset, int count) + { + Validate(offset, count); + Buffer.BlockCopy(_bytes, (int)offset, destination, destinationOffset, count); + return count; + } + + public void Dispose() => _bytes = Array.Empty(); + + private void Validate(long offset, int count) + { + if (offset < 0 || count < 0 || offset > _bytes.LongLength - count) + throw new ArgumentOutOfRangeException(nameof(offset)); + } } private void PumpMappedEntityFrame() diff --git a/packages/org.gamecult.eve.unity-scene/Runtime/CultMesh/GameCult.Eve.UnityScene.CultMesh.asmdef b/packages/org.gamecult.eve.unity-scene/Runtime/CultMesh/GameCult.Eve.UnityScene.CultMesh.asmdef index 72530cb..ba5091b 100644 --- a/packages/org.gamecult.eve.unity-scene/Runtime/CultMesh/GameCult.Eve.UnityScene.CultMesh.asmdef +++ b/packages/org.gamecult.eve.unity-scene/Runtime/CultMesh/GameCult.Eve.UnityScene.CultMesh.asmdef @@ -15,6 +15,7 @@ "GameCult.Caching.dll", "GameCult.Caching.MessagePack.dll", "GameCult.Mesh.dll", + "GameCult.Mesh.Quic.Native.dll", "GameCult.Networking.dll" ], "autoReferenced": true, diff --git a/packages/org.gamecult.eve.unity-scene/Tests/Editor/EveUnitySceneSurfaceLowererTests.cs b/packages/org.gamecult.eve.unity-scene/Tests/Editor/EveUnitySceneSurfaceLowererTests.cs index 05d134d..9a25ac0 100644 --- a/packages/org.gamecult.eve.unity-scene/Tests/Editor/EveUnitySceneSurfaceLowererTests.cs +++ b/packages/org.gamecult.eve.unity-scene/Tests/Editor/EveUnitySceneSurfaceLowererTests.cs @@ -115,6 +115,35 @@ namespace GameCult.Eve.UnityScene.Tests Assert.That(queued.GetType().GetProperty("Publication")!.GetValue(queued), Is.SameAs(publication)); } + [Test] + public void LiveTransportKeepsOnlyLatestPendingRealtimeGenerationPerBody() + { + using var transport = new EveUnityCultMeshLiveProviderTransport( + "test-replica.cc", + "cultnet://127.0.0.1:3075", + "provider", + "provider.pilot"); + var enqueue = typeof(EveUnityCultMeshLiveProviderTransport) + .GetMethod("EnqueueRealtimeEntityFrame", BindingFlags.NonPublic | BindingFlags.Instance)!; + var queue = (System.Collections.Concurrent.ConcurrentQueue) + typeof(EveUnityCultMeshLiveProviderTransport) + .GetField("_liveDocuments", BindingFlags.NonPublic | BindingFlags.Instance)! + .GetValue(transport)!; + var latest = (Dictionary) + typeof(EveUnityCultMeshLiveProviderTransport) + .GetField("_latestRealtimeFrames", BindingFlags.NonPublic | BindingFlags.Instance)! + .GetValue(transport)!; + + for (var sequence = 0; sequence < 100; sequence++) + enqueue.Invoke(transport, new object[] { RealtimeFrame("entities", 7, sequence) }); + enqueue.Invoke(transport, new object[] { RealtimeFrame("entities", 7, 50) }); + + Assert.That(queue.Count, Is.EqualTo(1), "superseded frames must not become render-thread debt"); + Assert.That(latest, Has.Count.EqualTo(1)); + Assert.That(latest["entities"].ProducerEpoch, Is.EqualTo(7)); + Assert.That(latest["entities"].Sequence, Is.EqualTo(99)); + } + [Test] public void LiveTransportReadsLaterMappedFramesWithoutNewControlDocuments() { @@ -2916,6 +2945,18 @@ namespace GameCult.Eve.UnityScene.Tests } }; + private static CultMeshRealtimeFrame RealtimeFrame(string bodyId, long epoch, long sequence) => + new CultMeshRealtimeFrame + { + ChannelId = "state", + BodyId = bodyId, + SchemaId = "test.entity.slab.v1", + ProducerEpoch = epoch, + Sequence = sequence, + Delivery = CultMeshRealtimeDelivery.LatestOnly, + Payload = new byte[4] + }; + private static EveProviderAdvertisementDocument ProviderAdvertisement( string providerId, string serviceId, diff --git a/packages/org.gamecult.eve.unity-scene/package.json b/packages/org.gamecult.eve.unity-scene/package.json index f255be1..bff0809 100644 --- a/packages/org.gamecult.eve.unity-scene/package.json +++ b/packages/org.gamecult.eve.unity-scene/package.json @@ -1,11 +1,11 @@ { "name": "org.gamecult.eve.unity-scene", "displayName": "Eve Unity Scene Lowering", - "version": "0.3.96", + "version": "0.3.97", "unity": "6000.0", "description": "Provider-agnostic Unity scene shell for gamecult.eve.surface.v1 interactive world surfaces.", "dependencies": { - "org.gamecult.cultlib": "1.0.35", + "org.gamecult.cultlib": "1.0.36", "org.gamecult.eve.plugin-fields": "0.2.3", "org.gamecult.eve.surface": "0.2.4", "com.unity.mathematics": "1.3.2", diff --git a/scripts/run-package-tests.ps1 b/scripts/run-package-tests.ps1 index dbf3962..b7f4b36 100644 --- a/scripts/run-package-tests.ps1 +++ b/scripts/run-package-tests.ps1 @@ -39,10 +39,21 @@ $arguments = @( "-assemblyNames", "GameCult.Eve.UnityScene.Tests", "-testResults", $resultsPath, "-logFile", $logPath ) -$process = Start-Process -FilePath $UnityExe -ArgumentList $arguments -Wait -PassThru -WindowStyle Hidden -if ($process.ExitCode -ne 0) { +$process = Start-Process -FilePath $UnityExe -ArgumentList $arguments -PassThru -WindowStyle Hidden +if (-not $process.WaitForExit([int][TimeSpan]::FromMinutes(10).TotalMilliseconds)) { + Stop-Process -Id $process.Id -Force -ErrorAction SilentlyContinue if (Test-Path -LiteralPath $logPath) { Get-Content -LiteralPath $logPath -Tail 120 } - throw "EveUnity package tests failed with exit code $($process.ExitCode)" + throw "EveUnity package tests timed out after 10 minutes (Unity PID $($process.Id))." +} +$unityExitCode = $process.ExitCode +if ($unityExitCode -ne 0) { + if (Test-Path -LiteralPath $logPath) { Get-Content -LiteralPath $logPath -Tail 120 } + throw "EveUnity package tests failed with exit code $unityExitCode" +} + +if (-not (Test-Path -LiteralPath $resultsPath)) { + if (Test-Path -LiteralPath $logPath) { Get-Content -LiteralPath $logPath -Tail 120 } + throw "EveUnity package tests produced no results: $resultsPath" } [xml]$results = Get-Content -LiteralPath $resultsPath -Raw