diff --git a/docs/authority-map.md b/docs/authority-map.md
index 4b25203..7a339b4 100644
--- a/docs/authority-map.md
+++ b/docs/authority-map.md
@@ -76,6 +76,10 @@ Unity code may not open provider snapshot sessions, schema clients, write
forwarders, subscriptions, or replica shards against a discovered physical
route. Reads, watches, content/body transfer, realtime frames, and command
submissions share the same stable-identity `CultMeshClient` owner.
+Remote bootstrap and asset materialization complete through `PrepareAsync`
+before the scene is mounted. `Connect`, `Mount`, input, refresh, and disposal
+cannot synchronously wait on CultMesh. A bounded command outbox owns delivery;
+only unsent continuous movement/look values may be replaced by a newer value.
Viewport-sized shader inputs are likewise presentation state. A provider may
relate a logical vector port to a logical native texture port through
@@ -98,8 +102,10 @@ modes into one camera opinion or infer a product camera from entity kind.
## Shared Paths
-Initial connect asks one rendezvous for Verses, selects a provider surface by
-semantic kind, then leases it through the selected stable provider identity.
+Initial preparation asks one rendezvous for Verses, selects an advertised
+authority runtime and provider surface by semantic kind, then leases it through
+the explicit Verse/authority-runtime target. The Eve provider ID remains a
+separate surface-owner identity.
Refresh, route rotation, reconnect, and terminal-receipt reconciliation remain
inside the same identity-owned session. Movement, focus, target, and action
input all become `gamecult.eve.command_invocation.v1` through the
diff --git a/packages/org.gamecult.eve.unity-scene/README.md b/packages/org.gamecult.eve.unity-scene/README.md
index 6edac13..e074c87 100644
--- a/packages/org.gamecult.eve.unity-scene/README.md
+++ b/packages/org.gamecult.eve.unity-scene/README.md
@@ -16,20 +16,23 @@ owned by their native lowerers and are not duplicated into empty UI elements.
networked client. Give it one Verse rendezvous endpoint and it discovers an
advertised `interactive-world`, constructs the generic CultMesh transport, and
provides every bootstrap port. Provider, surface, and Verse IDs are optional
-selection filters; they are not product knowledge required by the client.
-The selected provider ID remains the stable CultMesh endpoint identity while
-the Verse ID names the selected rule/progression world. Physical routes are
-resolved and re-resolved inside `CultMeshClient`; they are never
-published as provider selection state or cached as Unity authority. The optional
-local path is only an ephemeral content/body cache, not a replicated gameplay
-database.
+selection filters; they are not product knowledge required by the client. The
+selected Verse ID and its advertised authority-runtime ID form the stable
+CultMesh session target. The Eve provider ID separately names the UI owner found
+through that target. Physical routes are resolved and re-resolved inside
+`CultMeshClient`; they are never published as provider selection state or
+cached as Unity authority. The optional local path is only an ephemeral
+content/body cache, not a replicated gameplay database.
Await `PrepareAsync()` before mounting the bootstrap or reading provider ports.
-Discovery and session setup never run synchronously from Unity lifecycle or
-input getters, so an unavailable provider cannot stall the editor main thread.
-Live entity, field, and receipt polling likewise runs as one non-overlapping
-background operation. Unity's main thread only applies completed typed
-documents, keeping Play, Pause, inspection, and rendering responsive while a
-provider is slow or temporarily unreachable.
+It owns discovery, initial document reads, content transfer, asynchronous asset
+bundle loading, and live lease installation. Once preparation succeeds,
+`Mount()` and `Connect()` perform no remote work. Input submission only enters a
+bounded asynchronous outbox: one-shot commands retain their idempotency keys,
+while unsent movement and look values coalesce to the latest value per entity.
+Live entity, field, and receipt polling likewise remains non-blocking. Unity's
+main thread applies completed typed documents and assets; it never waits for a
+CultMesh task, keeping Play, Pause, input, inspection, and rendering responsive
+while a provider is slow or temporarily unreachable.
The package never imports provider product assemblies.
diff --git a/packages/org.gamecult.eve.unity-scene/Runtime/CultMesh/EveUnityCultMeshCommandOutbox.cs b/packages/org.gamecult.eve.unity-scene/Runtime/CultMesh/EveUnityCultMeshCommandOutbox.cs
new file mode 100644
index 0000000..6a93b30
--- /dev/null
+++ b/packages/org.gamecult.eve.unity-scene/Runtime/CultMesh/EveUnityCultMeshCommandOutbox.cs
@@ -0,0 +1,193 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using GameCult.Eve.Surface;
+
+#nullable enable
+
+namespace GameCult.Eve.UnityScene
+{
+ ///
+ /// Owns asynchronous command delivery for the CultMesh Unity transport.
+ /// The Unity input path only enqueues. One-shot commands retain their
+ /// idempotency key; continuous controls retain only the newest unsent value.
+ ///
+ internal sealed class EveUnityCultMeshCommandOutbox : IDisposable
+ {
+ private readonly object _gate = new object();
+ private readonly Queue _commands = new Queue();
+ private readonly Dictionary _continuous =
+ new Dictionary(StringComparer.Ordinal);
+ private readonly Func _send;
+ private readonly Func _continuousKey;
+ private readonly Action _discard;
+ private readonly int _commandCapacity;
+ private readonly int _continuousCapacity;
+ private readonly SemaphoreSlim _available = new SemaphoreSlim(0);
+ private readonly CancellationTokenSource _lifetime = new CancellationTokenSource();
+ private readonly Task _worker;
+ private bool _disposed;
+ private int _inFlight;
+
+ public EveUnityCultMeshCommandOutbox(
+ Func send,
+ Func continuousKey,
+ Action? discard = null,
+ int commandCapacity = 256,
+ int continuousCapacity = 32)
+ {
+ _send = send ?? throw new ArgumentNullException(nameof(send));
+ _continuousKey = continuousKey ?? throw new ArgumentNullException(nameof(continuousKey));
+ _discard = discard ?? (_ => { });
+ if (commandCapacity <= 0) throw new ArgumentOutOfRangeException(nameof(commandCapacity));
+ if (continuousCapacity <= 0) throw new ArgumentOutOfRangeException(nameof(continuousCapacity));
+ _commandCapacity = commandCapacity;
+ _continuousCapacity = continuousCapacity;
+ _worker = Task.Run(DeliverAsync);
+ }
+
+ internal int PendingCount
+ {
+ get
+ {
+ lock (_gate) return _commands.Count + _continuous.Count + _inFlight;
+ }
+ }
+
+ public void Enqueue(EveSurfaceCommandRequest request)
+ {
+ if (request == null) throw new ArgumentNullException(nameof(request));
+ if (string.IsNullOrWhiteSpace(request.CommandId))
+ throw new InvalidOperationException("Eve command invocations require an idempotency key.");
+
+ string? discarded = null;
+ var signal = false;
+ lock (_gate)
+ {
+ if (_disposed) throw new ObjectDisposedException(nameof(EveUnityCultMeshCommandOutbox));
+ var key = _continuousKey(request);
+ var item = new OutboxItem(request, string.IsNullOrWhiteSpace(key) ? null : key);
+ if (string.IsNullOrWhiteSpace(key))
+ {
+ if (_commands.Count >= _commandCapacity)
+ throw new InvalidOperationException(
+ $"The EveUnity command outbox reached its {_commandCapacity}-command bound.");
+ _commands.Enqueue(item);
+ signal = true;
+ }
+ else if (_continuous.TryGetValue(key, out var previous))
+ {
+ _continuous[key] = item;
+ discarded = previous.Request.CommandId;
+ }
+ else
+ {
+ if (_continuous.Count >= _continuousCapacity)
+ throw new InvalidOperationException(
+ $"The EveUnity continuous-control outbox reached its {_continuousCapacity}-binding bound.");
+ _continuous.Add(key, item);
+ signal = true;
+ }
+ }
+
+ if (!string.IsNullOrWhiteSpace(discarded)) _discard(discarded);
+ if (signal) _available.Release();
+ }
+
+ public void Dispose()
+ {
+ OutboxItem[] discarded;
+ lock (_gate)
+ {
+ if (_disposed) return;
+ _disposed = true;
+ discarded = _commands.Concat(_continuous.Values).ToArray();
+ _commands.Clear();
+ _continuous.Clear();
+ }
+ foreach (var item in discarded) _discard(item.Request.CommandId);
+ _lifetime.Cancel();
+ _available.Release();
+ }
+
+ private async Task DeliverAsync()
+ {
+ var cancellationToken = _lifetime.Token;
+ try
+ {
+ while (true)
+ {
+ await _available.WaitAsync(cancellationToken).ConfigureAwait(false);
+ var item = TakeNext();
+ if (item == null) continue;
+ try
+ {
+ var retryDelay = 50;
+ while (true)
+ {
+ try
+ {
+ await _send(item.Request, cancellationToken).ConfigureAwait(false);
+ break;
+ }
+ catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+ {
+ throw;
+ }
+ catch
+ {
+ if (WasSuperseded(item)) break;
+ await Task.Delay(retryDelay, cancellationToken).ConfigureAwait(false);
+ retryDelay = Math.Min(retryDelay * 2, 2000);
+ }
+ }
+ }
+ finally
+ {
+ lock (_gate) _inFlight--;
+ }
+ }
+ }
+ catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+ {
+ }
+ }
+
+ private OutboxItem? TakeNext()
+ {
+ lock (_gate)
+ {
+ if (_commands.Count > 0)
+ {
+ _inFlight++;
+ return _commands.Dequeue();
+ }
+ if (_continuous.Count == 0) return null;
+ var pair = _continuous.First();
+ _continuous.Remove(pair.Key);
+ _inFlight++;
+ return pair.Value;
+ }
+ }
+
+ private bool WasSuperseded(OutboxItem item)
+ {
+ if (item.ContinuousKey == null) return false;
+ lock (_gate) return _continuous.ContainsKey(item.ContinuousKey);
+ }
+
+ private sealed class OutboxItem
+ {
+ public OutboxItem(EveSurfaceCommandRequest request, string? continuousKey)
+ {
+ Request = request;
+ ContinuousKey = continuousKey;
+ }
+
+ public EveSurfaceCommandRequest Request { get; }
+ public string? ContinuousKey { get; }
+ }
+ }
+}
diff --git a/packages/org.gamecult.eve.unity-scene/Runtime/CultMesh/EveUnityCultMeshCommandOutbox.cs.meta b/packages/org.gamecult.eve.unity-scene/Runtime/CultMesh/EveUnityCultMeshCommandOutbox.cs.meta
new file mode 100644
index 0000000..a6c7e7a
--- /dev/null
+++ b/packages/org.gamecult.eve.unity-scene/Runtime/CultMesh/EveUnityCultMeshCommandOutbox.cs.meta
@@ -0,0 +1,3 @@
+fileFormatVersion: 2
+guid: 61d76e98f2054d86bf8e7480d4411c39
+timeCreated: 1787039650
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 26c22db..31922e8 100644
--- a/packages/org.gamecult.eve.unity-scene/Runtime/CultMesh/EveUnityCultMeshLiveProviderTransport.cs
+++ b/packages/org.gamecult.eve.unity-scene/Runtime/CultMesh/EveUnityCultMeshLiveProviderTransport.cs
@@ -45,6 +45,7 @@ namespace GameCult.Eve.UnityScene
new Dictionary(StringComparer.Ordinal);
private readonly HashSet _loadedBundleUris = new HashSet(StringComparer.Ordinal);
private readonly HashSet _loadingBundleUris = new HashSet(StringComparer.Ordinal);
+ private readonly HashSet _continuousCommandIds = new HashSet(StringComparer.Ordinal);
private readonly List _assetBodyLeases = new List();
private readonly Dictionary _renderChannelLayers = new Dictionary(StringComparer.Ordinal);
private readonly ConcurrentQueue _liveDocuments = new ConcurrentQueue();
@@ -57,6 +58,7 @@ namespace GameCult.Eve.UnityScene
new Dictionary(StringComparer.Ordinal);
private readonly List _documentLeases = new List();
private readonly List _documentWatches = new List();
+ private readonly CancellationTokenSource _lifetime = new CancellationTokenSource();
private CultMeshClient? _meshClient;
private bool _liveDocumentsReady;
private CultCache? _contentState;
@@ -73,6 +75,9 @@ namespace GameCult.Eve.UnityScene
private CancellationTokenSource? _realtimeLifetime;
private Task? _realtimePump;
private CultMeshRealtimeSession? _realtimeSession;
+ private EveUnityCultMeshCommandOutbox? _commandOutbox;
+ private EveAssetCatalogDocument? _pendingAssetCatalog;
+ private bool _assetCatalogUpdateRunning;
private long _lastQueuedEntityViewEpoch = -1;
private long _lastQueuedEntityViewSequence = -1;
private long _lastPresentedEntityViewEpoch = -1;
@@ -147,8 +152,7 @@ namespace GameCult.Eve.UnityScene
public void Connect()
{
- EnsureOpen();
- BootstrapIfRequired();
+ EnsurePrepared();
}
public void Disconnect()
@@ -157,32 +161,33 @@ namespace GameCult.Eve.UnityScene
public void Refresh()
{
- EnsureOpen();
- if (_bootstrapped)
- {
- PumpLiveEvents();
- return;
- }
-
- BootstrapIfRequired();
+ EnsurePrepared();
+ PumpLiveEvents();
}
- private void BootstrapIfRequired()
+ public async Task PrepareAsync(CancellationToken cancellationToken = default)
{
if (_bootstrapped)
return;
try
{
+ EnsureOpen();
var elapsed = Stopwatch.StartNew();
- ResolveAdvertisement(forceRefresh: true);
+ await ResolveAdvertisementAsync(forceRefresh: true, cancellationToken);
TraceStartup("advertisement", elapsed);
- RefreshSurface();
+ await RefreshSurfaceAsync(cancellationToken);
TraceStartup("surface", elapsed);
- RefreshAssetCatalog();
+ await RefreshAssetCatalogAsync(cancellationToken);
TraceStartup("asset-catalog-and-bundles", elapsed);
- EnsureLiveDocuments();
+ await PreloadAssetsAsync(cancellationToken);
+ TraceStartup("asset-preload", elapsed);
+ await EnsureLiveDocumentsAsync(cancellationToken);
TraceStartup("subscriptions", elapsed);
+ _commandOutbox = new EveUnityCultMeshCommandOutbox(
+ SendCommandAsync,
+ ContinuousCommandKey,
+ ForgetPendingCommand);
_bootstrapped = true;
}
catch (Exception error) when (error is IOException || error is SocketException || error is TimeoutException)
@@ -193,6 +198,13 @@ namespace GameCult.Eve.UnityScene
}
}
+ private void EnsurePrepared()
+ {
+ if (!_bootstrapped)
+ throw new InvalidOperationException(
+ "The EveUnity CultMesh transport is not prepared. Await PrepareAsync before mounting or connecting it.");
+ }
+
private static void TraceStartup(string phase, Stopwatch elapsed)
{
if (string.Equals(Environment.GetEnvironmentVariable("EVEUNITY_TRACE_STARTUP_PHASES"), "1", StringComparison.Ordinal))
@@ -234,7 +246,7 @@ namespace GameCult.Eve.UnityScene
else if (document is EveInputCapabilityDocument inputCapability)
CurrentInputCapability = inputCapability;
else if (document is EveAssetCatalogDocument assetCatalog)
- PublishAssetCatalog(assetCatalog);
+ QueueAssetCatalogUpdate(assetCatalog);
else if (document is EveCommandReceiptDocument receipt)
PublishReceipt(receipt);
else if (document is Exception error)
@@ -340,8 +352,7 @@ namespace GameCult.Eve.UnityScene
public void SubmitCommand(EveSurfaceCommandRequest request)
{
if (request == null) throw new ArgumentNullException(nameof(request));
- EnsureOpen();
- ResolveAdvertisement();
+ EnsurePrepared();
var interaction = RequireWorldInteraction();
if (string.IsNullOrWhiteSpace(interaction.CommandRecordRef))
throw new InvalidOperationException("The provider advertisement does not publish a command record reference.");
@@ -355,29 +366,48 @@ namespace GameCult.Eve.UnityScene
if (string.IsNullOrWhiteSpace(commandId))
throw new InvalidOperationException("Eve command invocations require an idempotency key.");
- _pendingCommandIds.Add(commandId);
- try
- {
- var recordKey = ChildRecordKey(interaction.CommandRecordRef, commandId);
- _meshClient!.SubmitDocumentAsync(
- _target,
- recordKey,
- request,
- _runtimeId,
- "eve-unity")
- .GetAwaiter()
- .GetResult();
- }
+ lock (_pendingCommandIds) _pendingCommandIds.Add(commandId);
+ try { _commandOutbox!.Enqueue(request); }
catch
{
- _pendingCommandIds.Remove(commandId);
+ ForgetPendingCommand(commandId);
throw;
}
}
+ private Task SendCommandAsync(EveSurfaceCommandRequest request, CancellationToken cancellationToken)
+ {
+ var interaction = RequireWorldInteraction();
+ var recordKey = ChildRecordKey(interaction.CommandRecordRef, request.CommandId);
+ return _meshClient!.SubmitDocumentAsync(
+ _target,
+ recordKey,
+ request,
+ _runtimeId,
+ "eve-unity",
+ cancellationToken);
+ }
+
+ private string? ContinuousCommandKey(EveSurfaceCommandRequest request)
+ {
+ if (!request.PayloadFields.TryGetValue("commandId", out var commandId) ||
+ !_continuousCommandIds.Contains(commandId))
+ return null;
+ request.PayloadFields.TryGetValue("entityId", out var entityId);
+ return commandId + "\u001f" + (entityId ?? "");
+ }
+
+ private void ForgetPendingCommand(string commandId)
+ {
+ lock (_pendingCommandIds) _pendingCommandIds.Remove(commandId);
+ }
+
public void Dispose()
{
_disposed = true;
+ _lifetime.Cancel();
+ _commandOutbox?.Dispose();
+ _commandOutbox = null;
foreach (var bundle in _assetBundles) bundle.Unload(unloadAllLoadedObjects: false);
_assetBundles.Clear();
foreach (var lease in _assetBodyLeases) lease.Dispose();
@@ -396,11 +426,6 @@ namespace GameCult.Eve.UnityScene
_realtimeLifetime?.Cancel();
_realtimeSession?.Dispose();
_realtimeSession = null;
- if (_realtimePump != null)
- {
- try { _realtimePump.GetAwaiter().GetResult(); }
- catch (OperationCanceledException) { }
- }
_realtimePump = null;
_realtimeLifetime?.Dispose();
_realtimeLifetime = null;
@@ -413,14 +438,18 @@ namespace GameCult.Eve.UnityScene
_assetBodyMappings = null;
_baseSurface = null;
_embeddedSurfaces.Clear();
- _pendingCommandIds.Clear();
- _publishedReceiptIds.Clear();
+ lock (_pendingCommandIds)
+ {
+ _pendingCommandIds.Clear();
+ _publishedReceiptIds.Clear();
+ }
+ _lifetime.Dispose();
}
public GameObject? ResolvePrefab(EveUnityPlayableWorldAssetBinding asset)
{
if (asset == null) throw new ArgumentNullException(nameof(asset));
- EnsureAssetLoaded(asset.AssetRef);
+ EnsureAssetPrepared(asset.AssetRef);
return _prefabs.TryGetValue(asset.AssetRef, out var prefab) ? prefab : null;
}
@@ -428,11 +457,19 @@ namespace GameCult.Eve.UnityScene
{
if (asset == null) throw new ArgumentNullException(nameof(asset));
if (assetType == null) throw new ArgumentNullException(nameof(assetType));
- EnsureAssetLoaded(asset.AssetRef);
+ EnsureAssetPrepared(asset.AssetRef);
return _nativeAssets.TryGetValue(asset.AssetRef, out var value) && assetType.IsInstanceOfType(value)
? value : null;
}
+ private void EnsureAssetPrepared(string assetRef)
+ {
+ EnsurePrepared();
+ if (_assetSelections.ContainsKey(assetRef) && !_nativeAssets.ContainsKey(assetRef))
+ throw new InvalidOperationException(
+ $"Provider asset '{assetRef}' was not materialized during asynchronous preparation.");
+ }
+
public bool TryResolveAssetMetadata(
EveUnityPlayableWorldAssetBinding asset,
out IReadOnlyDictionary metadata)
@@ -460,15 +497,16 @@ namespace GameCult.Eve.UnityScene
});
}
- private void ResolveAdvertisement(bool forceRefresh = false)
+ private async Task ResolveAdvertisementAsync(
+ bool forceRefresh = false,
+ CancellationToken cancellationToken = default)
{
if (!forceRefresh && _advertisement != null && _advertisedSurface != null)
return;
- using var advertisements = _meshClient!
- .LeaseCollectionAsync(_target)
- .GetAwaiter()
- .GetResult();
- _advertisement = advertisements.Handle.LatestAsync().GetAwaiter().GetResult().FirstOrDefault(document =>
+ using var lease = await _meshClient!
+ .LeaseCollectionAsync(_target, cancellationToken);
+ var documents = await lease.Handle.LatestAsync();
+ _advertisement = documents.FirstOrDefault(document =>
string.Equals(document.ProviderId, _providerId, StringComparison.Ordinal))
?? throw new InvalidOperationException($"Provider '{_providerId}' did not publish an Eve advertisement.");
_advertisedSurface = _advertisement.Surfaces.FirstOrDefault(surface =>
@@ -535,7 +573,7 @@ namespace GameCult.Eve.UnityScene
EveEntitySoaViewDocument document,
CultMeshBodyPublicationDocument publication)
{
- ResolveAdvertisement();
+ EnsurePrepared();
if (!string.Equals(document.ProviderId, _advertisement!.ProviderId, StringComparison.Ordinal))
throw new UnauthorizedAccessException(
$"Entity layout provider '{document.ProviderId}' does not match advertised Eve provider '{_advertisement.ProviderId}'.");
@@ -820,12 +858,10 @@ namespace GameCult.Eve.UnityScene
DateTimeOffset now) =>
publication.LivenessExpiresAtUnixMs > now.ToUnixTimeMilliseconds();
- private void RefreshSurface()
+ private async Task RefreshSurfaceAsync(CancellationToken cancellationToken)
{
- var surface = _meshClient!
- .ReadAsync(_target, _advertisedSurface!.RecordRef)
- .GetAwaiter()
- .GetResult();
+ var surface = await _meshClient!
+ .ReadAsync(_target, _advertisedSurface!.RecordRef, cancellationToken);
PublishBaseSurface(surface);
}
@@ -861,6 +897,13 @@ namespace GameCult.Eve.UnityScene
interaction.Ownership)),
_advertisedSurface.RecordRef,
surface.Version);
+ var lowered = new EveUnitySceneSurfaceLowerer()
+ .Lower(CurrentSurfaceDocument.SurfaceDocument, CurrentSurfaceDocument.AdvertisedSurface);
+ _continuousCommandIds.Clear();
+ if (!string.IsNullOrWhiteSpace(lowered.PlayableWorld?.MovementCommand))
+ _continuousCommandIds.Add(lowered.PlayableWorld.MovementCommand);
+ if (!string.IsNullOrWhiteSpace(lowered.PlayableWorld?.LookCommand))
+ _continuousCommandIds.Add(lowered.PlayableWorld.LookCommand);
SurfaceDocumentAvailable?.Invoke(CurrentSurfaceDocument);
}
@@ -908,13 +951,13 @@ namespace GameCult.Eve.UnityScene
component.Style);
}
- private void EnsureLiveDocuments()
+ private async Task EnsureLiveDocumentsAsync(CancellationToken cancellationToken)
{
if (_liveDocumentsReady) return;
DisposeLiveDocuments();
try
{
- OpenLiveDocuments();
+ await OpenLiveDocumentsAsync(cancellationToken);
_liveDocumentsReady = true;
}
catch
@@ -924,7 +967,7 @@ namespace GameCult.Eve.UnityScene
}
}
- private void OpenLiveDocuments()
+ private async Task OpenLiveDocumentsAsync(CancellationToken cancellationToken)
{
var lowered = new EveUnitySceneSurfaceLowerer()
.Lower(CurrentSurfaceDocument.SurfaceDocument, CurrentSurfaceDocument.AdvertisedSurface);
@@ -947,19 +990,23 @@ namespace GameCult.Eve.UnityScene
throw new InvalidOperationException(
"The playable world advertises an entity-view pointer without its logical body id.");
_bodyResolver ??= CreateBodyResolver();
- LeaseDocument(entityViewPointer, QueueEntityView);
- LeaseCollection(publication =>
+ await LeaseDocumentAsync(entityViewPointer, QueueEntityView, cancellationToken);
+ await LeaseCollectionAsync(publication =>
{
if (string.Equals(publication.BodyId, entityBodyId, StringComparison.Ordinal))
QueueBodyPublication(publication);
- });
+ }, cancellationToken);
}
foreach (var fieldRef in fieldRefs)
- LeaseDocument(fieldRef, fields => _liveDocuments.Enqueue(fields));
+ await LeaseDocumentAsync(
+ fieldRef,
+ fields => _liveDocuments.Enqueue(fields),
+ cancellationToken);
- LeaseDocument(
+ await LeaseDocumentAsync(
_advertisedSurface!.RecordRef,
surface => _liveDocuments.Enqueue(new PendingSurfaceDocument(_advertisedSurface.RecordRef, surface)),
+ cancellationToken,
publishInitial: false);
var embeddedSlots = EnumerateEmbeddedDocuments(CurrentSurfaceDocument.SurfaceDocument.Surface.Root)
.Where(slot => string.Equals(slot.SchemaId, EveSurfaceDocument.SchemaId, StringComparison.Ordinal))
@@ -969,29 +1016,33 @@ namespace GameCult.Eve.UnityScene
.ToArray();
foreach (var slot in embeddedSlots)
{
- LeaseDocument(
+ await LeaseDocumentAsync(
slot.DocumentId,
- surface => _liveDocuments.Enqueue(new PendingSurfaceDocument(slot.DocumentId, surface)));
+ surface => _liveDocuments.Enqueue(new PendingSurfaceDocument(slot.DocumentId, surface)),
+ cancellationToken);
}
var inputCapabilityRef = FindComponentProp(
CurrentSurfaceDocument.SurfaceDocument.Surface.Root,
"inputCapability");
if (!string.IsNullOrWhiteSpace(inputCapabilityRef))
{
- LeaseDocument(
+ await LeaseDocumentAsync(
inputCapabilityRef,
- document => _liveDocuments.Enqueue(document));
+ document => _liveDocuments.Enqueue(document),
+ cancellationToken);
}
var assetCatalogRef = RequireWorldInteraction().AssetManifestRecordRef;
if (!string.IsNullOrWhiteSpace(assetCatalogRef))
{
- LeaseDocument(
+ await LeaseDocumentAsync(
assetCatalogRef,
document => _liveDocuments.Enqueue(document),
+ cancellationToken,
publishInitial: false);
}
- LeaseCollection(
+ await LeaseCollectionAsync(
receipt => _liveDocuments.Enqueue(receipt),
+ cancellationToken,
includeInitialSnapshot: false);
}
@@ -1004,29 +1055,32 @@ namespace GameCult.Eve.UnityScene
_liveDocumentsReady = false;
}
- private void LeaseDocument(
+ private async Task LeaseDocumentAsync(
string recordKey,
Action publish,
+ CancellationToken cancellationToken,
bool publishInitial = true)
where TDocument : class
{
- var lease = _meshClient!.LeaseDocumentAsync(_target, recordKey)
- .GetAwaiter().GetResult();
+ var lease = await _meshClient!.LeaseDocumentAsync(_target, recordKey, cancellationToken);
_documentLeases.Add(lease);
- if (publishInitial) publish(lease.Handle.Latest());
+ if (publishInitial) publish(await lease.Handle.LatestAsync());
_documentWatches.Add(lease.Handle.Watch(publish));
}
- private void LeaseCollection(
+ private async Task LeaseCollectionAsync(
Action publish,
+ CancellationToken cancellationToken,
bool includeInitialSnapshot = true)
where TDocument : class
{
- var lease = _meshClient!.LeaseCollectionAsync(_target, includeInitialSnapshot)
- .GetAwaiter().GetResult();
+ var lease = await _meshClient!.LeaseCollectionAsync(
+ _target,
+ includeInitialSnapshot,
+ cancellationToken);
_documentLeases.Add(lease);
if (includeInitialSnapshot)
- foreach (var document in lease.Handle.LatestAsync().GetAwaiter().GetResult())
+ foreach (var document in await lease.Handle.LatestAsync())
publish(document);
_documentWatches.Add(lease.Handle.WatchChanges(change =>
{
@@ -1061,16 +1115,17 @@ namespace GameCult.Eve.UnityScene
Debug.Log($"EveUnity CultMesh hot state: {message}");
}
- private void RefreshAssetCatalog()
+ private async Task RefreshAssetCatalogAsync(CancellationToken cancellationToken)
{
var interaction = RequireWorldInteraction();
if (string.IsNullOrWhiteSpace(interaction.AssetManifestRecordRef))
return;
- var catalog = _meshClient!
- .ReadAsync(_target, interaction.AssetManifestRecordRef)
- .GetAwaiter()
- .GetResult();
+ var catalog = await _meshClient!
+ .ReadAsync(
+ _target,
+ interaction.AssetManifestRecordRef,
+ cancellationToken);
PublishAssetCatalog(catalog);
}
@@ -1136,14 +1191,49 @@ namespace GameCult.Eve.UnityScene
CurrentAssetCatalogVersion = catalog.Version;
}
- private void EnsureAssetLoaded(string assetRef)
+ private void QueueAssetCatalogUpdate(EveAssetCatalogDocument catalog)
{
- if (_nativeAssets.ContainsKey(assetRef) || !_assetSelections.TryGetValue(assetRef, out var selection))
- return;
- EnsureBundleLoaded(selection.Variant!.Uri);
+ _pendingAssetCatalog = catalog ?? throw new ArgumentNullException(nameof(catalog));
+ if (_assetCatalogUpdateRunning) return;
+ _assetCatalogUpdateRunning = true;
+ _ = DrainAssetCatalogUpdatesAsync();
}
- private void EnsureBundleLoaded(string uri)
+ private async Task DrainAssetCatalogUpdatesAsync()
+ {
+ try
+ {
+ while (!_lifetime.IsCancellationRequested)
+ {
+ var catalog = _pendingAssetCatalog;
+ _pendingAssetCatalog = null;
+ if (catalog == null) break;
+ PublishAssetCatalog(catalog);
+ await PreloadAssetsAsync(_lifetime.Token);
+ }
+ }
+ catch (OperationCanceledException) when (_lifetime.IsCancellationRequested)
+ {
+ }
+ catch (Exception error)
+ {
+ _liveDocuments.Enqueue(error);
+ }
+ finally
+ {
+ _assetCatalogUpdateRunning = false;
+ if (_pendingAssetCatalog != null && !_lifetime.IsCancellationRequested)
+ QueueAssetCatalogUpdate(_pendingAssetCatalog);
+ }
+ }
+
+ private async Task PreloadAssetsAsync(CancellationToken cancellationToken)
+ {
+ foreach (var uri in _bundleVariants.Keys.OrderBy(value => value, StringComparer.Ordinal).ToArray())
+ await EnsureBundleLoadedAsync(uri, cancellationToken);
+ }
+
+ private async Task EnsureBundleLoadedAsync(string uri, CancellationToken cancellationToken)
{
if (_loadedBundleUris.Contains(uri)) return;
if (!_bundleVariants.TryGetValue(uri, out var variant))
@@ -1154,14 +1244,13 @@ namespace GameCult.Eve.UnityScene
{
if (variant.Metadata.TryGetValue("unity.bundleDependencyUris", out var dependencies))
foreach (var dependency in dependencies.Split(new[] { ';' }, StringSplitOptions.RemoveEmptyEntries))
- EnsureBundleLoaded(dependency);
+ await EnsureBundleLoadedAsync(dependency, cancellationToken);
var groupElapsed = Stopwatch.StartNew();
- var descriptor = _meshClient!.ReadAsync(_target, uri)
- .GetAwaiter()
- .GetResult();
+ var descriptor = await _meshClient!
+ .ReadAsync(_target, uri, cancellationToken);
TraceStartup("asset-manifest", groupElapsed);
- var bundle = LoadVerifiedBundle(descriptor, variant);
+ var bundle = await LoadVerifiedBundleAsync(descriptor, variant, cancellationToken);
TraceStartup("asset-transfer-and-bundle-load", groupElapsed);
if (bundle == null)
throw new InvalidOperationException($"Unity could not load provider asset bundle '{uri}'.");
@@ -1172,7 +1261,7 @@ namespace GameCult.Eve.UnityScene
var bundleAssetName = ResolveBundleAssetName(bundleAssetNames, selection.Variant!.AssetKey);
var value = bundleAssetName == null
? null
- : LoadBundleAsset(bundle, bundleAssetName, selection.Asset.AssetKind);
+ : await LoadBundleAssetAsync(bundle, bundleAssetName, selection.Asset.AssetKind, cancellationToken);
if (value == null)
throw new InvalidOperationException(
$"Provider asset '{selection.Asset.AssetRef}' advertises missing Unity bundle asset " +
@@ -1196,11 +1285,17 @@ namespace GameCult.Eve.UnityScene
}
}
- private static UnityEngine.Object? LoadBundleAsset(AssetBundle bundle, string assetName, string assetKind)
+ private static async Task LoadBundleAssetAsync(
+ AssetBundle bundle,
+ string assetName,
+ string assetKind,
+ CancellationToken cancellationToken)
{
- return string.Equals(assetKind, "sprite", StringComparison.Ordinal)
- ? bundle.LoadAsset(assetName)
- : bundle.LoadAsset(assetName);
+ var request = string.Equals(assetKind, "sprite", StringComparison.Ordinal)
+ ? bundle.LoadAssetAsync(assetName)
+ : bundle.LoadAssetAsync(assetName);
+ await AwaitUnityAsync(request, cancellationToken);
+ return request.asset;
}
private sealed class SelectedAsset
@@ -1263,9 +1358,10 @@ namespace GameCult.Eve.UnityScene
private CultMeshContentTransferService ContentTransfer()
{
- ResolveAdvertisement();
if (_contentTransfer != null)
return _contentTransfer;
+ if (_advertisement == null)
+ throw new InvalidOperationException("The Eve provider advertisement must be prepared before content transfer.");
if (string.IsNullOrWhiteSpace(_advertisement!.ServiceId))
throw new InvalidOperationException($"Eve provider '{_advertisement.ProviderId}' has no service-instance identity for content transport.");
var cacheRoot = Environment.GetEnvironmentVariable("EVEUNITY_ASSET_CACHE_PATH");
@@ -1287,7 +1383,10 @@ namespace GameCult.Eve.UnityScene
return _contentTransfer;
}
- private AssetBundle? LoadVerifiedBundle(CultMeshCdnArtifactManifest manifest, EveAssetVariant variant)
+ private async Task LoadVerifiedBundleAsync(
+ CultMeshCdnArtifactManifest manifest,
+ EveAssetVariant variant,
+ CancellationToken cancellationToken)
{
var hash = NormalizeContentHash(variant.ContentHash);
if (manifest.SizeBytes != variant.SizeBytes)
@@ -1299,12 +1398,10 @@ namespace GameCult.Eve.UnityScene
var now = DateTimeOffset.UtcNow;
var network = NetworkArtifactDescriptor(manifest, now, TimeSpan.FromMinutes(5));
var elapsed = Stopwatch.StartNew();
- var mapped = ContentTransfer()
- .FetchMappedContentAsync(manifest, network, now, TimeSpan.FromMinutes(5))
- .GetAwaiter()
- .GetResult();
+ var mapped = await ContentTransfer()
+ .FetchMappedContentAsync(manifest, network, now, TimeSpan.FromMinutes(5), cancellationToken);
TraceStartup("content-materialization", elapsed);
- var request = new CultMeshBodyValidationRequest
+ var validationRequest = new CultMeshBodyValidationRequest
{
BodyId = network.BodyId,
SchemaId = network.SchemaId,
@@ -1319,33 +1416,27 @@ namespace GameCult.Eve.UnityScene
new ICultMeshBodyTransportAdapter[]
{
new CultMeshMappedBodyAdapter(_assetBodyMappings!),
- new CultMeshNetworkBodyAdapter(_ => ReadVerifiedArtifactBytes(manifest))
+ new CultMeshNetworkBodyAdapter(_ => File.ReadAllBytes(mapped.VerifiedPath))
},
descriptor => string.Equals(descriptor.BodyId, manifest.ArtifactId, StringComparison.Ordinal) &&
string.Equals(descriptor.SemanticHash, hash, StringComparison.Ordinal));
- var negotiated = transport.NegotiateReadOnly(mapped.Descriptor, network, request);
+ var negotiated = transport.NegotiateReadOnly(mapped.Descriptor, network, validationRequest);
CurrentAssetBodyTransportKind = negotiated.SelectedTransport;
_assetBodyLeases.Add(negotiated.Lease);
- if (negotiated.SelectedTransport == CultMeshBodyTransportKind.SharedFileMapping)
- {
- var bundle = AssetBundle.LoadFromFile(mapped.VerifiedPath);
- TraceStartup("asset-bundle-load-from-file", elapsed);
- return bundle;
- }
-
- if (negotiated.Lease.Descriptor.ByteSize > int.MaxValue)
- throw new InvalidOperationException("Unity cannot lower a network asset body larger than one managed byte array.");
- var bytes = new byte[checked((int)negotiated.Lease.Descriptor.ByteSize)];
- negotiated.Lease.CopyTo(0, bytes, 0, bytes.Length);
- var loaded = AssetBundle.LoadFromMemory(bytes);
- TraceStartup("asset-bundle-load-from-memory", elapsed);
- return loaded;
+ var bundleRequest = AssetBundle.LoadFromFileAsync(mapped.VerifiedPath);
+ await AwaitUnityAsync(bundleRequest, cancellationToken);
+ TraceStartup("asset-bundle-load-from-file", elapsed);
+ return bundleRequest.assetBundle;
}
- private byte[] ReadVerifiedArtifactBytes(CultMeshCdnArtifactManifest manifest)
+ private static async Task AwaitUnityAsync(AsyncOperation operation, CancellationToken cancellationToken)
{
- var path = ContentTransfer().FetchAsync(manifest).GetAwaiter().GetResult();
- return File.ReadAllBytes(path);
+ while (!operation.isDone)
+ {
+ cancellationToken.ThrowIfCancellationRequested();
+ await Task.Yield();
+ }
+ cancellationToken.ThrowIfCancellationRequested();
}
private static CultMeshBodyDescriptor NetworkArtifactDescriptor(
@@ -1375,8 +1466,12 @@ namespace GameCult.Eve.UnityScene
private void PublishReceipt(EveCommandReceiptDocument receipt)
{
- if (!_pendingCommandIds.Contains(receipt.CommandId) || !_publishedReceiptIds.Add(receipt.ReceiptId))
- return;
+ lock (_pendingCommandIds)
+ {
+ if (!_pendingCommandIds.Contains(receipt.CommandId) || !_publishedReceiptIds.Add(receipt.ReceiptId))
+ return;
+ _pendingCommandIds.Remove(receipt.CommandId);
+ }
CommandReceiptAvailable?.Invoke(new EveUnitySceneCommandReceipt(
receipt.ReceiptId,
receipt.Command,
@@ -1398,10 +1493,6 @@ namespace GameCult.Eve.UnityScene
receipt.Navigation.SurfaceId,
receipt.Navigation.SurfaceKind,
receipt.Navigation.RendezvousEndpoints)));
- if (string.Equals(receipt.State, "accepted", StringComparison.OrdinalIgnoreCase) ||
- string.Equals(receipt.State, "denied", StringComparison.OrdinalIgnoreCase) ||
- string.Equals(receipt.State, "reconciled", StringComparison.OrdinalIgnoreCase))
- _pendingCommandIds.Remove(receipt.CommandId);
}
private EveWorldInteractionAdvertisement RequireWorldInteraction()
diff --git a/packages/org.gamecult.eve.unity-scene/Runtime/CultMesh/EveUnityCultMeshPlayableWorldProvider.cs b/packages/org.gamecult.eve.unity-scene/Runtime/CultMesh/EveUnityCultMeshPlayableWorldProvider.cs
index 1e71df7..6f049bf 100644
--- a/packages/org.gamecult.eve.unity-scene/Runtime/CultMesh/EveUnityCultMeshPlayableWorldProvider.cs
+++ b/packages/org.gamecult.eve.unity-scene/Runtime/CultMesh/EveUnityCultMeshPlayableWorldProvider.cs
@@ -3,6 +3,7 @@ using System.IO;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
+using System.Threading;
using System.Threading.Tasks;
using Stopwatch = System.Diagnostics.Stopwatch;
using GameCult.Eve.Surface;
@@ -43,6 +44,7 @@ namespace GameCult.Eve.UnityScene
private EveUnityCultMeshLiveProviderTransport? _transport;
private EveUnitySceneLiveProviderBridge? _bridge;
private Task? _preparation;
+ private CancellationTokenSource? _preparationLifetime;
private readonly ConcurrentQueue _pendingEntityViews = new ConcurrentQueue();
private readonly ConcurrentQueue _pendingFields = new ConcurrentQueue();
@@ -109,7 +111,11 @@ namespace GameCult.Eve.UnityScene
if (_bridge != null)
return Task.CompletedTask;
if (_preparation == null || _preparation.IsCanceled || _preparation.IsFaulted)
- _preparation = PrepareTransportAsync();
+ {
+ _preparationLifetime?.Dispose();
+ _preparationLifetime = new CancellationTokenSource();
+ _preparation = PrepareTransportAsync(_preparationLifetime.Token);
+ }
return _preparation;
}
@@ -184,6 +190,7 @@ namespace GameCult.Eve.UnityScene
private void ReleaseTransport()
{
+ _preparationLifetime?.Cancel();
if (_bridge != null)
{
_bridge.DocumentAvailable -= ForwardDocument;
@@ -196,6 +203,8 @@ namespace GameCult.Eve.UnityScene
_bridge = null;
_transport = null;
_preparation = null;
+ _preparationLifetime?.Dispose();
+ _preparationLifetime = null;
Selection = null;
}
@@ -228,7 +237,7 @@ namespace GameCult.Eve.UnityScene
}
}
- private async Task PrepareTransportAsync()
+ private async Task PrepareTransportAsync(CancellationToken cancellationToken)
{
var elapsed = Stopwatch.StartNew();
if (_transport != null)
@@ -241,13 +250,14 @@ namespace GameCult.Eve.UnityScene
providerFilter,
surfaceFilter,
surfaceKind,
- verseFilter);
+ verseFilter,
+ cancellationToken);
TraceStartup($"discovery {elapsed.Elapsed.TotalMilliseconds:0.###}ms");
elapsed.Restart();
var resolvedCachePath = string.IsNullOrWhiteSpace(cacheDirectory)
? Path.Combine(Application.temporaryCachePath, $"eve-unity-{GetInstanceID()}")
: cacheDirectory;
- _transport = new EveUnityCultMeshLiveProviderTransport(
+ var transport = new EveUnityCultMeshLiveProviderTransport(
resolvedCachePath,
Selection.RendezvousEndpoint,
Selection.VerseId,
@@ -255,6 +265,17 @@ namespace GameCult.Eve.UnityScene
Selection.ProviderId,
Selection.SurfaceId,
runtimeId);
+ try
+ {
+ await transport.PrepareAsync(cancellationToken);
+ }
+ catch
+ {
+ transport.Dispose();
+ throw;
+ }
+ cancellationToken.ThrowIfCancellationRequested();
+ _transport = transport;
_transport.EntityViewAvailable += (view, lease) => _pendingEntityViews.Enqueue(new EntityViewLease(view, lease));
_transport.FieldsSplatsAvailable += fields => _pendingFields.Enqueue(fields);
_bridge = new EveUnitySceneLiveProviderBridge(_transport);
diff --git a/packages/org.gamecult.eve.unity-scene/Runtime/CultMesh/EveUnityCultMeshProviderDiscovery.cs b/packages/org.gamecult.eve.unity-scene/Runtime/CultMesh/EveUnityCultMeshProviderDiscovery.cs
index 74d3040..6628f28 100644
--- a/packages/org.gamecult.eve.unity-scene/Runtime/CultMesh/EveUnityCultMeshProviderDiscovery.cs
+++ b/packages/org.gamecult.eve.unity-scene/Runtime/CultMesh/EveUnityCultMeshProviderDiscovery.cs
@@ -1,6 +1,7 @@
using System;
using System.Collections.Generic;
using System.Linq;
+using System.Threading;
using System.Threading.Tasks;
using GameCult.Eve.Surface;
using GameCult.Mesh;
@@ -38,21 +39,13 @@ namespace GameCult.Eve.UnityScene
public sealed class EveUnityCultMeshProviderDiscovery
{
- public EveUnityCultMeshProviderSelection Discover(
- string rendezvousEndpoint,
- string providerId = "",
- string surfaceId = "",
- string surfaceKind = "interactive-world",
- string verseId = "") =>
- DiscoverAsync(rendezvousEndpoint, providerId, surfaceId, surfaceKind, verseId)
- .GetAwaiter().GetResult();
-
public async Task DiscoverAsync(
string rendezvousEndpoint,
string providerId = "",
string surfaceId = "",
string surfaceKind = "interactive-world",
- string verseId = "")
+ string verseId = "",
+ CancellationToken cancellationToken = default)
{
if (string.IsNullOrWhiteSpace(rendezvousEndpoint))
throw new ArgumentException("Rendezvous endpoint must be non-empty.", nameof(rendezvousEndpoint));
@@ -67,6 +60,7 @@ namespace GameCult.Eve.UnityScene
VerseIds = string.IsNullOrWhiteSpace(verseId) ? null : new[] { verseId },
TransportVersion = "cultmesh.v0"
});
+ cancellationToken.ThrowIfCancellationRequested();
}
catch (Exception error)
{
@@ -104,7 +98,7 @@ namespace GameCult.Eve.UnityScene
{
var target = new CultMeshSessionTarget(candidate.VerseId, authorityRuntimeId);
using var advertisementsLease = await mesh
- .LeaseCollectionAsync(target)
+ .LeaseCollectionAsync(target, cancellationToken)
.ConfigureAwait(false);
var advertisements = await advertisementsLease.Handle.LatestAsync().ConfigureAwait(false);
observed.AddRange(advertisements.Select(document =>
diff --git a/packages/org.gamecult.eve.unity-scene/Tests/Editor/EveUnityCultMeshAssetLookupTests.cs b/packages/org.gamecult.eve.unity-scene/Tests/Editor/EveUnityCultMeshAssetLookupTests.cs
index 5f1d489..1061353 100644
--- a/packages/org.gamecult.eve.unity-scene/Tests/Editor/EveUnityCultMeshAssetLookupTests.cs
+++ b/packages/org.gamecult.eve.unity-scene/Tests/Editor/EveUnityCultMeshAssetLookupTests.cs
@@ -5,8 +5,11 @@ using System.Diagnostics;
using System.Net;
using System.Net.Sockets;
using System.IO;
+using System.Threading;
+using System.Threading.Tasks;
using GameCult.Eve.Surface;
using GameCult.Eve.UnityScene;
+using GameCult.Mesh;
using NUnit.Framework;
using UnityEngine;
@@ -14,6 +17,79 @@ namespace GameCult.Eve.UnityScene.Tests
{
public sealed class EveUnityCultMeshAssetLookupTests
{
+ [Test]
+ public void ConnectCannotPerformRemoteWorkBeforeAsynchronousPreparation()
+ {
+ using var transport = new EveUnityCultMeshLiveProviderTransport(
+ "test-cache",
+ "cultnet+tcp://127.0.0.1:1",
+ "test.verse",
+ "test.runtime",
+ "test.provider",
+ "test.surface");
+
+ var elapsed = Stopwatch.StartNew();
+ var failure = Assert.Throws(() => transport.Connect());
+
+ StringAssert.Contains("Await PrepareAsync", failure!.Message);
+ Assert.That(elapsed.Elapsed, Is.LessThan(TimeSpan.FromMilliseconds(100)));
+ }
+
+ [Test]
+ public async Task CommandOutboxReturnsWhileDeliveryIsBlockedAndDeliversOneShotExactlyOnce()
+ {
+ var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+ var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+ var sends = 0;
+ using var outbox = new EveUnityCultMeshCommandOutbox(
+ async (_, cancellationToken) =>
+ {
+ Interlocked.Increment(ref sends);
+ entered.TrySetResult(true);
+ await AwaitOrCancel(release.Task, cancellationToken);
+ },
+ _ => null);
+
+ var elapsed = Stopwatch.StartNew();
+ outbox.Enqueue(Request("one-shot", "dock"));
+ Assert.That(elapsed.Elapsed, Is.LessThan(TimeSpan.FromMilliseconds(100)));
+ await RequireCompletion(entered.Task);
+ Assert.That(outbox.PendingCount, Is.EqualTo(1));
+
+ release.TrySetResult(true);
+ await WaitUntilAsync(() => outbox.PendingCount == 0);
+ Assert.That(sends, Is.EqualTo(1));
+ }
+
+ [Test]
+ public async Task CommandOutboxCoalescesBlockedContinuousInputToTheLatestValue()
+ {
+ var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+ var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+ var sent = new List();
+ using var outbox = new EveUnityCultMeshCommandOutbox(
+ async (request, cancellationToken) =>
+ {
+ lock (sent) sent.Add(request.CommandId);
+ entered.TrySetResult(true);
+ await AwaitOrCancel(release.Task, cancellationToken);
+ },
+ request => request.PayloadFields["commandId"] + "\u001f" + request.PayloadFields["entityId"]);
+
+ outbox.Enqueue(Request("move-0", "move"));
+ await RequireCompletion(entered.Task);
+ for (var index = 1; index <= 1000; index++)
+ outbox.Enqueue(Request("move-" + index, "move"));
+
+ Assert.That(outbox.PendingCount, Is.LessThanOrEqualTo(2));
+ release.TrySetResult(true);
+ await WaitUntilAsync(() => outbox.PendingCount == 0);
+ lock (sent)
+ {
+ CollectionAssert.AreEqual(new[] { "move-0", "move-1000" }, sent);
+ }
+ }
+
[Test]
public void ProviderSelectionCarriesStableIdentityWithoutExposingAPhysicalRoute()
{
@@ -31,6 +107,41 @@ namespace GameCult.Eve.UnityScene.Tests
Assert.That(typeof(EveUnityCultMeshProviderSelection).GetProperty("Endpoint"), Is.Null);
}
+ private static EveSurfaceCommandRequest Request(string id, string command) => new EveSurfaceCommandRequest(
+ "test.provider",
+ "test.surface",
+ CultMesh.OperationInvocation("test.command", idempotencyKey: id),
+ CultMesh.OperationPayload(new Dictionary
+ {
+ ["commandId"] = command,
+ ["entityId"] = "ship"
+ }),
+ DateTimeOffset.UtcNow,
+ "test-client");
+
+ private static async Task AwaitOrCancel(Task task, CancellationToken cancellationToken)
+ {
+ var cancelled = Task.Delay(Timeout.Infinite, cancellationToken);
+ if (await Task.WhenAny(task, cancelled) != task)
+ cancellationToken.ThrowIfCancellationRequested();
+ await task;
+ }
+
+ private static async Task RequireCompletion(Task task)
+ {
+ if (await Task.WhenAny(task, Task.Delay(TimeSpan.FromSeconds(2))) != task)
+ Assert.Fail("Timed out waiting for asynchronous command delivery.");
+ await task;
+ }
+
+ private static async Task WaitUntilAsync(Func predicate)
+ {
+ var deadline = DateTimeOffset.UtcNow + TimeSpan.FromSeconds(2);
+ while (!predicate() && DateTimeOffset.UtcNow < deadline)
+ await Task.Delay(10);
+ Assert.That(predicate(), Is.True, "Timed out waiting for the command outbox to drain.");
+ }
+
[Test]
public void LiveTransportOwnsOneCultMeshClientAndNoDirectSnapshotReplica()
{
diff --git a/packages/org.gamecult.eve.unity-scene/package.json b/packages/org.gamecult.eve.unity-scene/package.json
index b2c2a36..49bbacd 100644
--- a/packages/org.gamecult.eve.unity-scene/package.json
+++ b/packages/org.gamecult.eve.unity-scene/package.json
@@ -1,7 +1,7 @@
{
"name": "org.gamecult.eve.unity-scene",
"displayName": "Eve Unity Scene Lowering",
- "version": "0.3.106",
+ "version": "0.3.107",
"unity": "6000.0",
"description": "Provider-agnostic Unity scene shell for gamecult.eve.surface.v1 interactive world surfaces.",
"dependencies": {