Keep pending Eve commands alive until finality
This commit is contained in:
parent
039c979e72
commit
baf9df618e
3 changed files with 102 additions and 18 deletions
|
|
@ -31,7 +31,8 @@ namespace GameCult.Eve.UnityScene
|
|||
private readonly string _surfaceId;
|
||||
private readonly string _runtimeId;
|
||||
private readonly CultMeshAuthorityTrustPolicy _authorityTrust;
|
||||
private readonly HashSet<string> _pendingCommandIds = new HashSet<string>(StringComparer.Ordinal);
|
||||
private readonly Dictionary<string, EveSurfaceCommandRequest> _pendingCommands =
|
||||
new Dictionary<string, EveSurfaceCommandRequest>(StringComparer.Ordinal);
|
||||
private readonly Dictionary<string, Task<ReceiptSubscription>> _receiptSubscriptions =
|
||||
new Dictionary<string, Task<ReceiptSubscription>>(StringComparer.Ordinal);
|
||||
private readonly Dictionary<string, GameObject> _prefabs = new Dictionary<string, GameObject>(StringComparer.Ordinal);
|
||||
|
|
@ -371,7 +372,12 @@ namespace GameCult.Eve.UnityScene
|
|||
if (string.IsNullOrWhiteSpace(commandId))
|
||||
throw new InvalidOperationException("Eve command invocations require an idempotency key.");
|
||||
|
||||
lock (_pendingCommandIds) _pendingCommandIds.Add(commandId);
|
||||
lock (_pendingCommands)
|
||||
{
|
||||
if (_pendingCommands.ContainsKey(commandId))
|
||||
throw new InvalidOperationException($"Eve command '{commandId}' is already pending.");
|
||||
_pendingCommands.Add(commandId, request);
|
||||
}
|
||||
try { _commandOutbox!.Enqueue(request); }
|
||||
catch
|
||||
{
|
||||
|
|
@ -414,9 +420,9 @@ namespace GameCult.Eve.UnityScene
|
|||
private void ForgetPendingCommand(string commandId)
|
||||
{
|
||||
Task<ReceiptSubscription>? subscription = null;
|
||||
lock (_pendingCommandIds)
|
||||
lock (_pendingCommands)
|
||||
{
|
||||
_pendingCommandIds.Remove(commandId);
|
||||
_pendingCommands.Remove(commandId);
|
||||
if (_receiptSubscriptions.TryGetValue(commandId, out subscription))
|
||||
_receiptSubscriptions.Remove(commandId);
|
||||
}
|
||||
|
|
@ -459,9 +465,9 @@ namespace GameCult.Eve.UnityScene
|
|||
_assetBodyMappings = null;
|
||||
_baseSurface = null;
|
||||
_embeddedSurfaces.Clear();
|
||||
lock (_pendingCommandIds)
|
||||
lock (_pendingCommands)
|
||||
{
|
||||
_pendingCommandIds.Clear();
|
||||
_pendingCommands.Clear();
|
||||
foreach (var subscription in _receiptSubscriptions.Values)
|
||||
DisposeReceiptSubscription(subscription);
|
||||
_receiptSubscriptions.Clear();
|
||||
|
|
@ -1080,7 +1086,7 @@ namespace GameCult.Eve.UnityScene
|
|||
CancellationToken cancellationToken)
|
||||
{
|
||||
Task<ReceiptSubscription> subscription;
|
||||
lock (_pendingCommandIds)
|
||||
lock (_pendingCommands)
|
||||
{
|
||||
if (_receiptSubscriptions.TryGetValue(commandId, out subscription!)) return subscription;
|
||||
subscription = OpenReceiptSubscriptionAsync(commandId, cancellationToken);
|
||||
|
|
@ -1096,7 +1102,7 @@ namespace GameCult.Eve.UnityScene
|
|||
try { return await subscription.ConfigureAwait(false); }
|
||||
catch
|
||||
{
|
||||
lock (_pendingCommandIds)
|
||||
lock (_pendingCommands)
|
||||
{
|
||||
if (_receiptSubscriptions.TryGetValue(commandId, out var current) &&
|
||||
ReferenceEquals(current, subscription))
|
||||
|
|
@ -1134,11 +1140,13 @@ namespace GameCult.Eve.UnityScene
|
|||
|
||||
private void QueueReceipt(string expectedCommandId, EveCommandReceiptDocument receipt)
|
||||
{
|
||||
if (receipt == null ||
|
||||
!string.Equals(receipt.CommandId, expectedCommandId, StringComparison.Ordinal) ||
|
||||
!string.Equals(receipt.ProviderId, _providerId, StringComparison.Ordinal) ||
|
||||
!string.Equals(receipt.SurfaceId, _surfaceId, StringComparison.Ordinal))
|
||||
return;
|
||||
if (receipt == null) return;
|
||||
lock (_pendingCommands)
|
||||
{
|
||||
if (!_pendingCommands.TryGetValue(expectedCommandId, out var request) ||
|
||||
!ReceiptMatches(request, receipt))
|
||||
return;
|
||||
}
|
||||
_liveDocuments.Enqueue(receipt);
|
||||
}
|
||||
|
||||
|
|
@ -1574,13 +1582,19 @@ namespace GameCult.Eve.UnityScene
|
|||
|
||||
private void PublishReceipt(EveCommandReceiptDocument receipt)
|
||||
{
|
||||
lock (_pendingCommandIds)
|
||||
var terminal = IsTerminalReceiptState(receipt.State);
|
||||
lock (_pendingCommands)
|
||||
{
|
||||
if (!_pendingCommandIds.Remove(receipt.CommandId))
|
||||
if (!_pendingCommands.TryGetValue(receipt.CommandId, out var request) ||
|
||||
!ReceiptMatches(request, receipt))
|
||||
return;
|
||||
if (terminal) _pendingCommands.Remove(receipt.CommandId);
|
||||
}
|
||||
if (terminal)
|
||||
{
|
||||
_commandOutbox?.Acknowledge(receipt.CommandId);
|
||||
ForgetPendingCommand(receipt.CommandId);
|
||||
}
|
||||
_commandOutbox?.Acknowledge(receipt.CommandId);
|
||||
ForgetPendingCommand(receipt.CommandId);
|
||||
CommandReceiptAvailable?.Invoke(new EveUnitySceneCommandReceipt(
|
||||
receipt.ReceiptId,
|
||||
receipt.Command,
|
||||
|
|
@ -1605,6 +1619,24 @@ namespace GameCult.Eve.UnityScene
|
|||
receipt.Navigation.AuthorityRuntimeId)));
|
||||
}
|
||||
|
||||
private bool ReceiptMatches(EveSurfaceCommandRequest request, EveCommandReceiptDocument receipt) =>
|
||||
string.Equals(receipt.Schema, EveCommandReceiptDocument.SchemaId, StringComparison.Ordinal) &&
|
||||
string.Equals(receipt.CommandId, request.CommandId, StringComparison.Ordinal) &&
|
||||
string.Equals(receipt.Command, request.Command, StringComparison.Ordinal) &&
|
||||
string.Equals(receipt.ProviderId, request.ProviderId, StringComparison.Ordinal) &&
|
||||
string.Equals(receipt.SurfaceId, request.SurfaceId, StringComparison.Ordinal) &&
|
||||
string.Equals(receipt.Authority, _target.AuthorityRuntimeId, StringComparison.Ordinal) &&
|
||||
IsKnownReceiptState(receipt.State);
|
||||
|
||||
private static bool IsKnownReceiptState(string state) =>
|
||||
string.Equals(state, "pending", StringComparison.OrdinalIgnoreCase) ||
|
||||
IsTerminalReceiptState(state);
|
||||
|
||||
private static bool IsTerminalReceiptState(string state) =>
|
||||
string.Equals(state, "accepted", StringComparison.OrdinalIgnoreCase) ||
|
||||
string.Equals(state, "denied", StringComparison.OrdinalIgnoreCase) ||
|
||||
string.Equals(state, "reconciled", StringComparison.OrdinalIgnoreCase);
|
||||
|
||||
private sealed class ReceiptSubscription : IDisposable
|
||||
{
|
||||
private IDisposable? _lease;
|
||||
|
|
|
|||
|
|
@ -2,6 +2,7 @@ using System;
|
|||
using System.Reflection;
|
||||
using System.Collections.Generic;
|
||||
using System.Diagnostics;
|
||||
using System.Linq;
|
||||
using System.Net;
|
||||
using System.Net.Sockets;
|
||||
using System.IO;
|
||||
|
|
@ -151,6 +152,41 @@ namespace GameCult.Eve.UnityScene.Tests
|
|||
await WaitUntilAsync(() => outbox.PendingCount == 0);
|
||||
}
|
||||
|
||||
[Test]
|
||||
public void PendingReceiptIsObservedWithoutRetiringTheCommand()
|
||||
{
|
||||
using var transport = new EveUnityCultMeshLiveProviderTransport(
|
||||
"test-cache",
|
||||
"cultnet+tcp://127.0.0.1:1",
|
||||
"test.verse",
|
||||
"test.runtime",
|
||||
"test.provider",
|
||||
"test.surface");
|
||||
var request = Request("launch-1", "launch");
|
||||
var pendingField = typeof(EveUnityCultMeshLiveProviderTransport).GetField(
|
||||
"_pendingCommands",
|
||||
BindingFlags.NonPublic | BindingFlags.Instance);
|
||||
var publish = typeof(EveUnityCultMeshLiveProviderTransport).GetMethod(
|
||||
"PublishReceipt",
|
||||
BindingFlags.NonPublic | BindingFlags.Instance);
|
||||
Assert.That(pendingField, Is.Not.Null);
|
||||
Assert.That(publish, Is.Not.Null);
|
||||
var pending = (IDictionary<string, EveSurfaceCommandRequest>)pendingField!.GetValue(transport)!;
|
||||
pending.Add(request.CommandId, request);
|
||||
var observed = new List<EveUnitySceneCommandReceipt>();
|
||||
transport.CommandReceiptAvailable += observed.Add;
|
||||
|
||||
publish!.Invoke(transport, new object[] { Receipt(request, "pending", "receipt-pending") });
|
||||
|
||||
Assert.That(pending.ContainsKey(request.CommandId), Is.True);
|
||||
Assert.That(observed.Select(value => value.State), Is.EqualTo(new[] { "pending" }));
|
||||
|
||||
publish.Invoke(transport, new object[] { Receipt(request, "accepted", "receipt-accepted") });
|
||||
|
||||
Assert.That(pending.ContainsKey(request.CommandId), Is.False);
|
||||
Assert.That(observed.Select(value => value.State), Is.EqualTo(new[] { "pending", "accepted" }));
|
||||
}
|
||||
|
||||
[Test]
|
||||
public void ProviderSelectionCarriesStableIdentityWithoutExposingAPhysicalRoute()
|
||||
{
|
||||
|
|
@ -198,6 +234,22 @@ namespace GameCult.Eve.UnityScene.Tests
|
|||
DateTimeOffset.UtcNow,
|
||||
"test-client");
|
||||
|
||||
private static EveCommandReceiptDocument Receipt(
|
||||
EveSurfaceCommandRequest request,
|
||||
string state,
|
||||
string receiptId) => new EveCommandReceiptDocument(
|
||||
receiptId,
|
||||
request.CommandId,
|
||||
request.Command,
|
||||
state,
|
||||
"AetheriaEve",
|
||||
"test.runtime",
|
||||
request.ProviderId,
|
||||
request.SurfaceId,
|
||||
"",
|
||||
DateTimeOffset.UtcNow.ToString("O"),
|
||||
1);
|
||||
|
||||
private static async Task AwaitOrCancel(Task task, CancellationToken cancellationToken)
|
||||
{
|
||||
var cancelled = Task.Delay(Timeout.Infinite, cancellationToken);
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
{
|
||||
"name": "org.gamecult.eve.unity-scene",
|
||||
"displayName": "Eve Unity Scene Lowering",
|
||||
"version": "0.3.121",
|
||||
"version": "0.3.122",
|
||||
"unity": "6000.0",
|
||||
"description": "Provider-agnostic Unity scene shell for gamecult.eve.surface.v1 interactive world surfaces.",
|
||||
"dependencies": {
|
||||
|
|
|
|||
Reference in a new issue