diff --git a/docs/authority-map.md b/docs/authority-map.md index 7a339b4..3be5f6c 100644 --- a/docs/authority-map.md +++ b/docs/authority-map.md @@ -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 diff --git a/packages/org.gamecult.eve.unity-scene/README.md b/packages/org.gamecult.eve.unity-scene/README.md index e074c87..c54381c 100644 --- a/packages/org.gamecult.eve.unity-scene/README.md +++ b/packages/org.gamecult.eve.unity-scene/README.md @@ -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 diff --git a/packages/org.gamecult.eve.unity-scene/Runtime/CultMesh/EveUnityCultMeshCommandOutbox.cs b/packages/org.gamecult.eve.unity-scene/Runtime/CultMesh/EveUnityCultMeshCommandOutbox.cs index 6a93b30..f8c89fa 100644 --- a/packages/org.gamecult.eve.unity-scene/Runtime/CultMesh/EveUnityCultMeshCommandOutbox.cs +++ b/packages/org.gamecult.eve.unity-scene/Runtime/CultMesh/EveUnityCultMeshCommandOutbox.cs @@ -10,14 +10,25 @@ using GameCult.Eve.Surface; 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. + /// 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. /// internal sealed class EveUnityCultMeshCommandOutbox : IDisposable { + internal enum DeliveryState + { + Queued, + Sending, + AwaitingCanonicalReceipt + } + private readonly object _gate = new object(); - private readonly Queue _commands = new Queue(); + private readonly List _items = new List(); + private readonly Dictionary _byCommandId = + new Dictionary(StringComparer.Ordinal); private readonly Dictionary _continuous = new Dictionary(StringComparer.Ordinal); private readonly Func _send; @@ -25,18 +36,24 @@ namespace GameCult.Eve.UnityScene private readonly Action _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 send, Func continuousKey, Action? 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(); + } + + /// Completes a command only when its canonical provider receipt arrives. + 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; } } } } 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 31922e8..0ea02b6 100644 --- a/packages/org.gamecult.eve.unity-scene/Runtime/CultMesh/EveUnityCultMeshLiveProviderTransport.cs +++ b/packages/org.gamecult.eve.unity-scene/Runtime/CultMesh/EveUnityCultMeshLiveProviderTransport.cs @@ -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, 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 1061353..8e70be8 100644 --- a/packages/org.gamecult.eve.unity-scene/Tests/Editor/EveUnityCultMeshAssetLookupTests.cs +++ b/packages/org.gamecult.eve.unity-scene/Tests/Editor/EveUnityCultMeshAssetLookupTests.cs @@ -36,7 +36,7 @@ namespace GameCult.Eve.UnityScene.Tests } [Test] - public async Task CommandOutboxReturnsWhileDeliveryIsBlockedAndDeliversOneShotExactlyOnce() + public async Task CommandOutboxCompletesOnlyAfterCanonicalReceipt() { var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); var release = new TaskCompletionSource(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(); + 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() { diff --git a/packages/org.gamecult.eve.unity-scene/package.json b/packages/org.gamecult.eve.unity-scene/package.json index 41e1b2e..8d7b92c 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.108", + "version": "0.3.109", "unity": "6000.0", "description": "Provider-agnostic Unity scene shell for gamecult.eve.surface.v1 interactive world surfaces.", "dependencies": { diff --git a/scripts/run-package-tests.ps1 b/scripts/run-package-tests.ps1 index acd058d..6f4357a 100644 --- a/scripts/run-package-tests.ps1 +++ b/scripts/run-package-tests.ps1 @@ -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)) {