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

Keep Unity commands pending until provider receipts

Browse source
This commit is contained in:
Meta Vens 2026-08-18 17:47:34 +02:00
commit 1354dac447
7 changed files with 258 additions and 78 deletions
Download patch file Download diff file Expand all files Collapse all files

5
docs/authority-map.md
View file

@ -79,7 +79,10 @@ 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.
continuous movement/look values may be replaced by a newer value, but transport
send completion cannot retire any command. Only the provider's canonical
receipt retires its idempotency key. Unacknowledged work is retried fairly; it
cannot become client-authored success or starve unrelated commands.
Viewport-sized shader inputs are likewise presentation state. A provider may
relate a logical vector port to a logical native texture port through

6
packages/org.gamecult.eve.unity-scene/README.md
View file

@ -28,7 +28,11 @@ 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.
while movement and look values coalesce to the latest value per entity. A
successful send is not completion: the command remains pending and retries with
the same idempotency key until its canonical provider receipt arrives. Failed
or unacknowledged commands rotate behind other due work, so one poison command
cannot starve the outbox.
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

241
packages/org.gamecult.eve.unity-scene/Runtime/CultMesh/EveUnityCultMeshCommandOutbox.cs
View file

@ -10,14 +10,25 @@ using GameCult.Eve.Surface;
namespace GameCult.Eve.UnityScene
{
/// <summary>
/// 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.
/// Owns receipt-backed asynchronous command delivery for the CultMesh Unity
/// transport. A successful transport send only moves a command into
/// AwaitingCanonicalReceipt. The provider receipt is the sole completion
/// signal. Retries retain the original idempotency key, and one failed
/// command cannot monopolize delivery of the rest of the bounded outbox.
/// </summary>
internal sealed class EveUnityCultMeshCommandOutbox : IDisposable
{
internal enum DeliveryState
{
Queued,
Sending,
AwaitingCanonicalReceipt
}
private readonly object _gate = new object();
private readonly Queue<OutboxItem> _commands = new Queue<OutboxItem>();
private readonly List<OutboxItem> _items = new List<OutboxItem>();
private readonly Dictionary<string, OutboxItem> _byCommandId =
new Dictionary<string, OutboxItem>(StringComparer.Ordinal);
private readonly Dictionary<string, OutboxItem> _continuous =
new Dictionary<string, OutboxItem>(StringComparer.Ordinal);
private readonly Func<EveSurfaceCommandRequest, CancellationToken, Task> _send;
@ -25,18 +36,24 @@ namespace GameCult.Eve.UnityScene
private readonly Action<string> _discard;
private readonly int _commandCapacity;
private readonly int _continuousCapacity;
private readonly SemaphoreSlim _available = new SemaphoreSlim(0);
private readonly TimeSpan _retryDelay;
private readonly TimeSpan _maximumRetryDelay;
private readonly TimeSpan _sendAttemptTimeout;
private readonly SemaphoreSlim _changed = new SemaphoreSlim(0);
private readonly CancellationTokenSource _lifetime = new CancellationTokenSource();
private readonly Task _worker;
private bool _disposed;
private int _inFlight;
private long _sequence;
public EveUnityCultMeshCommandOutbox(
Func<EveSurfaceCommandRequest, CancellationToken, Task> send,
Func<EveSurfaceCommandRequest, string?> continuousKey,
Action<string>? discard = null,
int commandCapacity = 256,
int continuousCapacity = 32)
int continuousCapacity = 32,
TimeSpan? retryDelay = null,
TimeSpan? maximumRetryDelay = null,
TimeSpan? sendAttemptTimeout = null)
{
_send = send ?? throw new ArgumentNullException(nameof(send));
_continuousKey = continuousKey ?? throw new ArgumentNullException(nameof(continuousKey));
@ -45,15 +62,24 @@ namespace GameCult.Eve.UnityScene
if (continuousCapacity <= 0) throw new ArgumentOutOfRangeException(nameof(continuousCapacity));
_commandCapacity = commandCapacity;
_continuousCapacity = continuousCapacity;
_retryDelay = retryDelay ?? TimeSpan.FromMilliseconds(100);
_maximumRetryDelay = maximumRetryDelay ?? TimeSpan.FromSeconds(2);
_sendAttemptTimeout = sendAttemptTimeout ?? TimeSpan.FromSeconds(2);
if (_retryDelay <= TimeSpan.Zero) throw new ArgumentOutOfRangeException(nameof(retryDelay));
if (_maximumRetryDelay < _retryDelay) throw new ArgumentOutOfRangeException(nameof(maximumRetryDelay));
if (_sendAttemptTimeout <= TimeSpan.Zero) throw new ArgumentOutOfRangeException(nameof(sendAttemptTimeout));
_worker = Task.Run(DeliverAsync);
}
internal int PendingCount
{
get
{
lock (_gate) return _commands.Count + _continuous.Count + _inFlight;
}
get { lock (_gate) return _items.Count; }
}
internal DeliveryState? StateOf(string commandId)
{
lock (_gate)
return _byCommandId.TryGetValue(commandId, out var item) ? item.State : null;
}
public void Enqueue(EveSurfaceCommandRequest request)
@ -63,37 +89,52 @@ namespace GameCult.Eve.UnityScene
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));
if (_byCommandId.ContainsKey(request.CommandId))
throw new InvalidOperationException($"Eve command '{request.CommandId}' is already pending.");
var key = _continuousKey(request);
var item = new OutboxItem(request, string.IsNullOrWhiteSpace(key) ? null : key);
if (string.IsNullOrWhiteSpace(key))
key = string.IsNullOrWhiteSpace(key) ? null : key;
if (key == null)
{
if (_commands.Count >= _commandCapacity)
if (_items.Count(item => item.ContinuousKey == null) >= _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;
RemoveLocked(previous);
discarded = previous.Request.CommandId;
}
else
else if (_continuous.Count >= _continuousCapacity)
{
if (_continuous.Count >= _continuousCapacity)
throw new InvalidOperationException(
$"The EveUnity continuous-control outbox reached its {_continuousCapacity}-binding bound.");
_continuous.Add(key, item);
signal = true;
throw new InvalidOperationException(
$"The EveUnity continuous-control outbox reached its {_continuousCapacity}-binding bound.");
}
var item = new OutboxItem(request, key, ++_sequence);
_items.Add(item);
_byCommandId.Add(request.CommandId, item);
if (key != null) _continuous[key] = item;
}
if (!string.IsNullOrWhiteSpace(discarded)) _discard(discarded);
if (signal) _available.Release();
if (discarded != null) _discard(discarded);
_changed.Release();
}
/// <summary>Completes a command only when its canonical provider receipt arrives.</summary>
public bool Acknowledge(string commandId)
{
if (string.IsNullOrWhiteSpace(commandId)) return false;
lock (_gate)
{
if (!_byCommandId.TryGetValue(commandId, out var item)) return false;
RemoveLocked(item);
}
_changed.Release();
return true;
}
public void Dispose()
@ -103,13 +144,14 @@ namespace GameCult.Eve.UnityScene
{
if (_disposed) return;
_disposed = true;
discarded = _commands.Concat(_continuous.Values).ToArray();
_commands.Clear();
discarded = _items.ToArray();
_items.Clear();
_byCommandId.Clear();
_continuous.Clear();
}
foreach (var item in discarded) _discard(item.Request.CommandId);
_lifetime.Cancel();
_available.Release();
_changed.Release();
}
private async Task DeliverAsync()
@ -119,35 +161,13 @@ namespace GameCult.Eve.UnityScene
{
while (true)
{
await _available.WaitAsync(cancellationToken).ConfigureAwait(false);
var item = TakeNext();
if (item == null) continue;
try
var selection = TakeNext();
if (selection.Item == null)
{
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--;
await WaitForChangeOrDueAsync(selection.Delay, cancellationToken).ConfigureAwait(false);
continue;
}
await DeliverOnceAsync(selection.Item, cancellationToken).ConfigureAwait(false);
}
}
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
@ -155,39 +175,122 @@ namespace GameCult.Eve.UnityScene
}
}
private OutboxItem? TakeNext()
private async Task DeliverOnceAsync(OutboxItem item, CancellationToken cancellationToken)
{
var sent = false;
using (var attempt = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken))
{
attempt.CancelAfter(_sendAttemptTimeout);
try
{
await _send(item.Request, attempt.Token).ConfigureAwait(false);
sent = true;
}
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
{
throw;
}
catch
{
// The same idempotency key is retried after other due items.
}
}
lock (_gate)
{
if (!_byCommandId.TryGetValue(item.Request.CommandId, out var current) ||
!ReferenceEquals(item, current))
return;
item.Attempts++;
item.State = sent ? DeliveryState.AwaitingCanonicalReceipt : DeliveryState.Queued;
item.NextAttemptAt = DateTimeOffset.UtcNow + RetryDelay(item.Attempts);
item.Sequence = ++_sequence;
}
_changed.Release();
}
private Selection TakeNext()
{
lock (_gate)
{
if (_commands.Count > 0)
var now = DateTimeOffset.UtcNow;
var item = _items
.Where(candidate => candidate.State != DeliveryState.Sending && candidate.NextAttemptAt <= now)
.OrderBy(candidate => candidate.Sequence)
.FirstOrDefault();
if (item != null)
{
_inFlight++;
return _commands.Dequeue();
item.State = DeliveryState.Sending;
return new Selection(item, null);
}
if (_continuous.Count == 0) return null;
var pair = _continuous.First();
_continuous.Remove(pair.Key);
_inFlight++;
return pair.Value;
var next = _items
.Where(candidate => candidate.State != DeliveryState.Sending)
.Select(candidate => (DateTimeOffset?)candidate.NextAttemptAt)
.OrderBy(value => value)
.FirstOrDefault();
return new Selection(null, next.HasValue ? Max(TimeSpan.Zero, next.Value - now) : null);
}
}
private bool WasSuperseded(OutboxItem item)
private async Task WaitForChangeOrDueAsync(TimeSpan? delay, CancellationToken cancellationToken)
{
if (item.ContinuousKey == null) return false;
lock (_gate) return _continuous.ContainsKey(item.ContinuousKey);
if (!delay.HasValue)
{
await _changed.WaitAsync(cancellationToken).ConfigureAwait(false);
return;
}
var milliseconds = (int)Math.Min(int.MaxValue, Math.Max(1, delay.Value.TotalMilliseconds));
await _changed.WaitAsync(milliseconds, cancellationToken).ConfigureAwait(false);
}
private TimeSpan RetryDelay(int attempts)
{
var multiplier = Math.Pow(2, Math.Min(attempts - 1, 10));
return TimeSpan.FromMilliseconds(Math.Min(
_maximumRetryDelay.TotalMilliseconds,
_retryDelay.TotalMilliseconds * multiplier));
}
private void RemoveLocked(OutboxItem item)
{
_items.Remove(item);
_byCommandId.Remove(item.Request.CommandId);
if (item.ContinuousKey != null &&
_continuous.TryGetValue(item.ContinuousKey, out var current) &&
ReferenceEquals(item, current))
_continuous.Remove(item.ContinuousKey);
}
private static TimeSpan Max(TimeSpan left, TimeSpan right) => left >= right ? left : right;
private sealed class OutboxItem
{
public OutboxItem(EveSurfaceCommandRequest request, string? continuousKey)
public OutboxItem(EveSurfaceCommandRequest request, string? continuousKey, long sequence)
{
Request = request;
ContinuousKey = continuousKey;
Sequence = sequence;
NextAttemptAt = DateTimeOffset.MinValue;
}
public EveSurfaceCommandRequest Request { get; }
public string? ContinuousKey { get; }
public DeliveryState State { get; set; }
public DateTimeOffset NextAttemptAt { get; set; }
public int Attempts { get; set; }
public long Sequence { get; set; }
}
private readonly struct Selection
{
public Selection(OutboxItem? item, TimeSpan? delay)
{
Item = item;
Delay = delay;
}
public OutboxItem? Item { get; }
public TimeSpan? Delay { get; }
}
}
}

1
packages/org.gamecult.eve.unity-scene/Runtime/CultMesh/EveUnityCultMeshLiveProviderTransport.cs
View file

@ -1472,6 +1472,7 @@ namespace GameCult.Eve.UnityScene
return;
_pendingCommandIds.Remove(receipt.CommandId);
}
_commandOutbox?.Acknowledge(receipt.CommandId);
CommandReceiptAvailable?.Invoke(new EveUnitySceneCommandReceipt(
receipt.ReceiptId,
receipt.Command,

69
packages/org.gamecult.eve.unity-scene/Tests/Editor/EveUnityCultMeshAssetLookupTests.cs
View file

@ -36,7 +36,7 @@ namespace GameCult.Eve.UnityScene.Tests
}
[Test]
public async Task CommandOutboxReturnsWhileDeliveryIsBlockedAndDeliversOneShotExactlyOnce()
public async Task CommandOutboxCompletesOnlyAfterCanonicalReceipt()
{
var entered = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
var release = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
@ -48,7 +48,8 @@ namespace GameCult.Eve.UnityScene.Tests
entered.TrySetResult(true);
await AwaitOrCancel(release.Task, cancellationToken);
},
_ => null);
_ => null,
retryDelay: TimeSpan.FromMilliseconds(250));
var elapsed = Stopwatch.StartNew();
outbox.Enqueue(Request("one-shot", "dock"));
@ -57,8 +58,13 @@ namespace GameCult.Eve.UnityScene.Tests
Assert.That(outbox.PendingCount, Is.EqualTo(1));
release.TrySetResult(true);
await WaitUntilAsync(() => outbox.PendingCount == 0);
await WaitUntilAsync(() => outbox.StateOf("one-shot") ==
EveUnityCultMeshCommandOutbox.DeliveryState.AwaitingCanonicalReceipt);
Assert.That(outbox.PendingCount, Is.EqualTo(1));
Assert.That(sends, Is.EqualTo(1));
Assert.That(outbox.Acknowledge("one-shot"), Is.True);
Assert.That(outbox.Acknowledge("one-shot"), Is.False);
await WaitUntilAsync(() => outbox.PendingCount == 0);
}
[Test]
@ -74,7 +80,8 @@ namespace GameCult.Eve.UnityScene.Tests
entered.TrySetResult(true);
await AwaitOrCancel(release.Task, cancellationToken);
},
request => request.PayloadFields["commandId"] + "\u001f" + request.PayloadFields["entityId"]);
request => request.PayloadFields["commandId"] + "\u001f" + request.PayloadFields["entityId"],
retryDelay: TimeSpan.FromMilliseconds(250));
outbox.Enqueue(Request("move-0", "move"));
await RequireCompletion(entered.Task);
@ -83,6 +90,9 @@ namespace GameCult.Eve.UnityScene.Tests
Assert.That(outbox.PendingCount, Is.LessThanOrEqualTo(2));
release.TrySetResult(true);
await WaitUntilAsync(() => outbox.StateOf("move-1000") ==
EveUnityCultMeshCommandOutbox.DeliveryState.AwaitingCanonicalReceipt);
Assert.That(outbox.Acknowledge("move-1000"), Is.True);
await WaitUntilAsync(() => outbox.PendingCount == 0);
lock (sent)
{
@ -90,6 +100,57 @@ namespace GameCult.Eve.UnityScene.Tests
}
}
[Test]
public async Task CommandOutboxRetriesSameIdWithoutLettingPoisonCommandStarveLaterWork()
{
var attempts = new List<string>();
using var outbox = new EveUnityCultMeshCommandOutbox(
(request, _) =>
{
lock (attempts) attempts.Add(request.CommandId);
if (request.CommandId == "poison") throw new IOException("route unavailable");
return Task.CompletedTask;
},
_ => null,
retryDelay: TimeSpan.FromMilliseconds(20),
maximumRetryDelay: TimeSpan.FromMilliseconds(40));
outbox.Enqueue(Request("poison", "dock"));
outbox.Enqueue(Request("healthy", "undock"));
await WaitUntilAsync(() => outbox.StateOf("healthy") ==
EveUnityCultMeshCommandOutbox.DeliveryState.AwaitingCanonicalReceipt);
Assert.That(outbox.Acknowledge("healthy"), Is.True);
lock (attempts)
{
Assert.That(attempts.IndexOf("healthy"), Is.GreaterThanOrEqualTo(0));
Assert.That(attempts.FindAll(value => value == "poison"), Is.Not.Empty);
}
Assert.That(outbox.PendingCount, Is.EqualTo(1));
}
[Test]
public async Task CommandOutboxRetransmitsUnacknowledgedCommandWithOriginalId()
{
var attempts = 0;
using var outbox = new EveUnityCultMeshCommandOutbox(
(request, _) =>
{
Assert.That(request.CommandId, Is.EqualTo("retry-me"));
Interlocked.Increment(ref attempts);
return Task.CompletedTask;
},
_ => null,
retryDelay: TimeSpan.FromMilliseconds(20),
maximumRetryDelay: TimeSpan.FromMilliseconds(20));
outbox.Enqueue(Request("retry-me", "dock"));
await WaitUntilAsync(() => Volatile.Read(ref attempts) >= 2);
Assert.That(outbox.PendingCount, Is.EqualTo(1));
Assert.That(outbox.Acknowledge("retry-me"), Is.True);
await WaitUntilAsync(() => outbox.PendingCount == 0);
}
[Test]
public void ProviderSelectionCarriesStableIdentityWithoutExposingAPhysicalRoute()
{

2
packages/org.gamecult.eve.unity-scene/package.json
View file

@ -1,7 +1,7 @@
{
"name": "org.gamecult.eve.unity-scene",
"displayName": "Eve Unity Scene Lowering",
"version": "0.3.108",
"version": "0.3.109",
"unity": "6000.0",
"description": "Provider-agnostic Unity scene shell for gamecult.eve.surface.v1 interactive world surfaces.",
"dependencies": {

12
scripts/run-package-tests.ps1
View file

@ -3,6 +3,8 @@ param(
[string] $CultLibRoot = "",
[string] $EveRoot = "",
[string] $EvePluginsRoot = "",
[string] $NuGetConfig = "",
[switch] $SkipCultLibBuild,
[string] $ProjectRoot = "TestProject",
[string] $OutputRoot = "artifacts\package-tests"
)
@ -33,8 +35,14 @@ foreach ($required in @(
if (-not (Test-Path -LiteralPath $required)) { throw "Required EveUnity package-test path not found: $required" }
}
powershell -ExecutionPolicy Bypass -File $cultLibBuilder
if ($LASTEXITCODE -ne 0) { throw "CultLib Unity package build failed with exit code $LASTEXITCODE" }
if (-not $SkipCultLibBuild) {
$cultLibBuildArguments = @("-ExecutionPolicy", "Bypass", "-File", $cultLibBuilder)
if (-not [string]::IsNullOrWhiteSpace($NuGetConfig)) {
$cultLibBuildArguments += @("-NuGetConfig", [IO.Path]::GetFullPath($NuGetConfig))
}
& powershell @cultLibBuildArguments
if ($LASTEXITCODE -ne 0) { throw "CultLib Unity package build failed with exit code $LASTEXITCODE" }
}
$builtCultLibPackage = Join-Path $CultLibRoot "artifacts\unity\org.gamecult.cultlib"
if (-not (Test-Path -LiteralPath $builtCultLibPackage)) {