From e5b41d0925a7ea97606db3cda1f454d772c5e8b7 Mon Sep 17 00:00:00 2001 From: Eric Sun <141227631+EricSun0218@users.noreply.github.com> Date: Sat, 15 Aug 2026 23:50:39 +0800 Subject: [PATCH] Add bounded engine-thread action queue --- CHANGELOG.md | 4 +- README.md | 2 +- README.zh-CN.md | 2 +- docs/features.md | 2 +- docs/getting-started.md | 23 + src/OpenGameAgent/QueuedGameActionHandler.cs | 515 ++++++++++++++++++ .../PublicApiCompatibilityTests.cs | 2 +- .../QueuedGameActionHandlerTests.cs | 376 +++++++++++++ 8 files changed, 921 insertions(+), 5 deletions(-) create mode 100644 src/OpenGameAgent/QueuedGameActionHandler.cs create mode 100644 tests/OpenGameAgent.Tests/QueuedGameActionHandlerTests.cs diff --git a/CHANGELOG.md b/CHANGELOG.md index 0808392..c6fc9a8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,7 +2,9 @@ ## Unreleased -No changes yet. +- Add `QueuedGameActionHandler`, a bounded engine-thread action handoff with FIFO pumping, + queued cancellation, active-action settlement, shutdown cleanup, and main-thread recovery over + the existing durable action journal and receipt protocol. ## 0.3.0-alpha.2 diff --git a/README.md b/README.md index 2bd3a3e..c663955 100644 --- a/README.md +++ b/README.md @@ -123,7 +123,7 @@ Read [Architecture](docs/architecture.md) for the ownership and failure boundari | Image input | PNG/JPEG/WebP/GIF admission, immutable content-addressed storage, reference-only transcripts, capability preflight, tool-result images, and authorized server retrieval | | Extension API | Immutable builder; prompt/context/tool/skill/route/workflow/hook/provider/service registration; typed lifecycle events and channels; namespaced persistent state | | Official extensions | Tool policy and search, structured player questions/recommended replies, goals, host-verified ordered task plans with durable pause/resume, memory, artifacts, knowledge, delegation, tracing, and durable parallel workflow graphs | -| World primitives | Durable actions, resumable workflows, memories, skills, signals, game-time schedules, actor mailboxes with batch read-only pending status | +| World primitives | Durable actions, bounded engine-thread action handoff, resumable workflows, memories, skills, signals, game-time schedules, actor mailboxes with batch read-only pending status | | Models and auth | Bundled capability/context/reasoning/cost directory, dynamic refresh, API-key/environment/stored/OAuth/local auth, developer-hosted short-lived credential gateway | | External tools | Lazy on-demand search/describe/call by default; explicit direct exposure for small trusted catalogs | | Portable plugins | [Agent Plugins 1.0.0](docs/agent-plugins.md) `plugin.json`, immediate-child `SKILL.md` discovery, MCP stdio/Streamable HTTP, client namespaces, containment, and component-level failure isolation | diff --git a/README.zh-CN.md b/README.zh-CN.md index c105487..c12d88d 100644 --- a/README.zh-CN.md +++ b/README.zh-CN.md @@ -121,7 +121,7 @@ GameAgentRuntime | 图片输入 | PNG/JPEG/WebP/GIF 准入、不可变内容寻址存储、仅引用会话、模型能力预检、工具结果图片与授权服务端读取 | | 扩展 API | 不可变构建器；提示词/上下文/工具/Skills/路由/Workflow/Hooks/提供方/服务注册；类型化生命周期事件与通道；命名空间持久状态 | | 官方扩展 | 工具策略与搜索、玩家结构化提问/推荐回复、目标、支持持久暂停/恢复且由宿主校验证据的有序任务清单、记忆、产物、外部知识、委派、追踪和可持久并行工作流图 | -| 世界原语 | 可恢复动作、可续跑 Workflow、记忆、Skills、信号、游戏时间调度、支持批量只读待处理状态的角色邮箱 | +| 世界原语 | 可恢复动作、有界引擎线程动作交接、可续跑 Workflow、记忆、Skills、信号、游戏时间调度、支持批量只读待处理状态的角色邮箱 | | 模型与认证 | 内置模型能力/上下文/推理级别/成本目录、动态刷新、API Key/环境/存储/OAuth/本地认证、开发者托管短期凭证网关 | | 外部工具 | 默认按需搜索/描述/调用；小型可信目录可显式选择原生直连暴露 | | 可移植插件 | [Agent Plugins 1.0.0](docs/agent-plugins.md) `plugin.json`、直接子目录 `SKILL.md` 发现、MCP stdio/Streamable HTTP、客户端命名空间、路径限制与组件级故障隔离 | diff --git a/docs/features.md b/docs/features.md index fc30301..4a5b4dc 100644 --- a/docs/features.md +++ b/docs/features.md @@ -64,7 +64,7 @@ This page maps product needs to the smallest reusable OpenGameAgent primitive. | --- | --- | | Mutate game state through a typed tool | `GameActionTool` | | Avoid repeating uncertain writes | `DurableGameActionDispatcher`, `IGameActionJournal` | -| Execute on engine main thread | implement `IGameActionHandler` by queueing into the engine, then await the receipt | +| Execute on an engine-owned main thread | wrap the authoritative `IGameActionHandler` in `QueuedGameActionHandler`, then call `Pump` from the engine thread | | Store long-term NPC facts/events | `IGameMemoryStore`, `GameMemory` | | Apply custom semantic ranking | `IGameMemoryRanker`, `RankedGameMemoryStore` | | Add local or remote vector embeddings and hybrid recall | `IMemoryEmbeddingProvider`, `VectorMemoryStore` | diff --git a/docs/getting-started.md b/docs/getting-started.md index a33957b..819204f 100644 --- a/docs/getting-started.md +++ b/docs/getting-started.md @@ -153,6 +153,29 @@ ValueTask> Tools(GameInput input, CancellationToken _) `IGameActionHandler.ExecuteAsync` must recheck visibility, permission, resources, expected revision, and all game rules. Return `Rejected` for a legal request that cannot commit. Implement `RecoverAsync` using the game's operation ledger or transaction log. +When the model loop runs off the engine thread, wrap that authoritative handler in +`QueuedGameActionHandler` and pump it from the engine thread. The durable dispatcher remains the +only action journal; the queue is a bounded, process-local handoff: + +```csharp +var engineActions = new QueuedGameActionHandler( + new AuthoritativeGameActionHandler(world), + maximumPendingActions: 256, + maximumActiveActions: 16); +var dispatcher = new DurableGameActionDispatcher(actionJournal, engineActions); + +// Unity Update, Godot _Process, or the equivalent host-owned main-thread callback. +engineActions.Pump(maximumWorkItems: 16); +``` + +The first `Pump` call binds the instance to that managed thread. Cancellation removes an action +only while it is still queued. Once the host starts an action, caller timeout no longer cancels the +mutation blindly; its receipt is allowed to settle. `Stop` rejects new work and faults queued work, +while `DisposeAsync` additionally waits for already-started work. The wrapped authoritative handler +must validate `GenerationId` against the currently loaded save/world generation because only the +game knows which generation is active. Pending durable journal entries are recovered through the +same pump after restart. + ## Select routes The default route is intentionally simple: diff --git a/src/OpenGameAgent/QueuedGameActionHandler.cs b/src/OpenGameAgent/QueuedGameActionHandler.cs new file mode 100644 index 0000000..d699c54 --- /dev/null +++ b/src/OpenGameAgent/QueuedGameActionHandler.cs @@ -0,0 +1,515 @@ +namespace OpenGameAgent; + +/// +/// Marshals authoritative game actions onto a host-owned pump thread while preserving the +/// durable action protocol implemented by . +/// +/// +/// Call from the engine thread that is allowed to touch game state. The +/// first call binds the handler to that managed thread. Caller cancellation removes work only +/// while it is still queued; after an action starts, its outcome is allowed to settle so that a +/// caller timeout cannot silently cancel a world mutation. +/// +public sealed class QueuedGameActionHandler : IGameActionHandler, IDisposable, IAsyncDisposable +{ + private const int MaximumSupportedCapacity = 1_000_000; + + private readonly object _gate = new(); + private readonly IGameActionHandler _innerHandler; + private readonly LinkedList _queue = new(); + private readonly TaskCompletionSource _stoppedCompletion = + new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly int _maximumPendingActions; + private readonly int _maximumActiveActions; + private bool _accepting = true; + private int _activeCount; + private int _pumpThreadId; + private int _pumping; + + public QueuedGameActionHandler( + IGameActionHandler innerHandler, + int maximumPendingActions = 1_024, + int maximumActiveActions = 64) + { + _innerHandler = innerHandler ?? throw new ArgumentNullException(nameof(innerHandler)); + if (maximumPendingActions <= 0 || maximumPendingActions > MaximumSupportedCapacity) + { + throw new ArgumentOutOfRangeException(nameof(maximumPendingActions)); + } + + if (maximumActiveActions <= 0 || maximumActiveActions > MaximumSupportedCapacity) + { + throw new ArgumentOutOfRangeException(nameof(maximumActiveActions)); + } + + _maximumPendingActions = maximumPendingActions; + _maximumActiveActions = maximumActiveActions; + } + + public int MaximumPendingActions => _maximumPendingActions; + + public int MaximumActiveActions => _maximumActiveActions; + + public bool IsAccepting + { + get + { + lock (_gate) + { + return _accepting; + } + } + } + + public int PendingCount + { + get + { + lock (_gate) + { + return _queue.Count; + } + } + } + + public int ActiveCount + { + get + { + lock (_gate) + { + return _activeCount; + } + } + } + + public ValueTask ExecuteAsync( + GameActionIntent intent, + CancellationToken cancellationToken) + { + var completion = Enqueue(intent, WorkKind.Execute, cancellationToken); + return AwaitRequiredReceiptAsync(completion); + } + + public ValueTask RecoverAsync( + GameActionIntent intent, + CancellationToken cancellationToken) => + new(Enqueue(intent, WorkKind.Recover, cancellationToken)); + + /// + /// Starts up to queued operations on the calling thread. + /// Incomplete asynchronous handlers remain bounded by . + /// + /// The number of work items started by this call. + public int Pump(int maximumWorkItems) + { + if (maximumWorkItems <= 0 || maximumWorkItems > MaximumSupportedCapacity) + { + throw new ArgumentOutOfRangeException(nameof(maximumWorkItems)); + } + + BindPumpThread(); + if (Interlocked.CompareExchange(ref _pumping, 1, 0) != 0) + { + throw new InvalidOperationException("The game action pump cannot run concurrently or reentrantly."); + } + + try + { + var started = 0; + while (started < maximumWorkItems) + { + WorkItem? item; + CancellationTokenRegistration cancellationRegistration; + lock (_gate) + { + if (_queue.First is null || _activeCount >= _maximumActiveActions) + { + break; + } + + item = _queue.First.Value; + _queue.RemoveFirst(); + item.Node = null; + item.State = WorkState.Started; + _activeCount++; + cancellationRegistration = item.CancellationRegistration; + item.CancellationRegistration = default; + } + + cancellationRegistration.Dispose(); + Start(item); + started++; + } + + return started; + } + finally + { + Volatile.Write(ref _pumping, 0); + } + } + + /// + /// Rejects new work and faults work that has not started. Active operations are not canceled. + /// + public void Stop() + { + List? pending = null; + var stopped = false; + lock (_gate) + { + if (!_accepting) + { + return; + } + + _accepting = false; + if (_queue.Count > 0) + { + pending = new List(_queue.Count); + while (_queue.First is { } node) + { + _queue.RemoveFirst(); + node.Value.Node = null; + node.Value.State = WorkState.Stopped; + pending.Add(node.Value); + } + } + + stopped = _activeCount == 0; + } + + if (pending is not null) + { + foreach (var item in pending) + { + item.CancellationRegistration.Dispose(); + item.CancellationRegistration = default; + item.Completion.TrySetException( + new ObjectDisposedException(nameof(QueuedGameActionHandler), "The game action queue has stopped.")); + } + } + + if (stopped) + { + _stoppedCompletion.TrySetResult(null); + } + } + + public void Dispose() + { + Stop(); + GC.SuppressFinalize(this); + } + + public async ValueTask DisposeAsync() + { + Stop(); + await _stoppedCompletion.Task.ConfigureAwait(false); + GC.SuppressFinalize(this); + } + + private Task Enqueue( + GameActionIntent intent, + WorkKind kind, + CancellationToken cancellationToken) + { + if (intent is null) + { + throw new ArgumentNullException(nameof(intent)); + } + + if (cancellationToken.IsCancellationRequested) + { + return Task.FromCanceled(cancellationToken); + } + + var item = new WorkItem(this, intent, kind); + lock (_gate) + { + if (!_accepting) + { + throw new ObjectDisposedException(nameof(QueuedGameActionHandler), "The game action queue has stopped."); + } + + if (_queue.Count >= _maximumPendingActions) + { + throw new GameRuntimeLimitException( + nameof(_maximumPendingActions), + "The game action queue reached its pending capacity."); + } + + item.Node = _queue.AddLast(item); + if (cancellationToken.CanBeCanceled) + { + var state = new CancellationState(item, cancellationToken); + item.CancellationRegistration = cancellationToken.Register( + static callbackState => + { + var cancellation = (CancellationState)callbackState!; + cancellation.Item.Owner.CancelQueued(cancellation.Item, cancellation.Token); + }, + state); + + if (item.State != WorkState.Queued) + { + QueueCancellationRegistrationRelease(item); + } + } + } + + return item.Completion.Task; + } + + private void CancelQueued(WorkItem item, CancellationToken cancellationToken) + { + lock (_gate) + { + if (item.State != WorkState.Queued || item.Node is null) + { + return; + } + + _queue.Remove(item.Node); + item.Node = null; + item.State = WorkState.Canceled; + } + + item.Completion.TrySetCanceled(cancellationToken); + QueueCancellationRegistrationRelease(item); + } + + private static void QueueCancellationRegistrationRelease(WorkItem item) + { + ThreadPool.QueueUserWorkItem( + static state => + { + var workItem = (WorkItem)state!; + workItem.Owner.ReleaseCancellationRegistration(workItem); + }, + item); + } + + private void ReleaseCancellationRegistration(WorkItem item) + { + CancellationTokenRegistration registration; + lock (_gate) + { + registration = item.CancellationRegistration; + item.CancellationRegistration = default; + } + + registration.Dispose(); + } + + private void BindPumpThread() + { + var currentThreadId = Environment.CurrentManagedThreadId; + var pumpThreadId = Volatile.Read(ref _pumpThreadId); + if (pumpThreadId == 0) + { + pumpThreadId = Interlocked.CompareExchange(ref _pumpThreadId, currentThreadId, 0); + if (pumpThreadId == 0) + { + pumpThreadId = currentThreadId; + } + } + + if (pumpThreadId != currentThreadId) + { + throw new InvalidOperationException("The game action pump must always run on the thread that first called Pump."); + } + } + + private void Start(WorkItem item) + { + if (item.Kind == WorkKind.Execute) + { + StartExecute(item); + return; + } + + StartRecover(item); + } + + private void StartExecute(WorkItem item) + { + try + { + var operation = _innerHandler.ExecuteAsync(item.Intent, CancellationToken.None); + if (operation.IsCompleted) + { + Complete(item, operation.GetAwaiter().GetResult(), exception: null, canceled: false); + } + else + { + _ = ObserveExecuteAsync(item, operation); + } + } + catch (OperationCanceledException exception) + { + Complete(item, receipt: null, exception, canceled: true); + } + catch (Exception exception) + { + Complete(item, receipt: null, exception, canceled: false); + } + } + + private void StartRecover(WorkItem item) + { + try + { + var operation = _innerHandler.RecoverAsync(item.Intent, CancellationToken.None); + if (operation.IsCompleted) + { + Complete(item, operation.GetAwaiter().GetResult(), exception: null, canceled: false); + } + else + { + _ = ObserveRecoverAsync(item, operation); + } + } + catch (OperationCanceledException exception) + { + Complete(item, receipt: null, exception, canceled: true); + } + catch (Exception exception) + { + Complete(item, receipt: null, exception, canceled: false); + } + } + + private async Task ObserveExecuteAsync(WorkItem item, ValueTask operation) + { + try + { + var receipt = await operation.ConfigureAwait(false); + Complete(item, receipt, exception: null, canceled: false); + } + catch (OperationCanceledException exception) + { + Complete(item, receipt: null, exception, canceled: true); + } + catch (Exception exception) + { + Complete(item, receipt: null, exception, canceled: false); + } + } + + private async Task ObserveRecoverAsync(WorkItem item, ValueTask operation) + { + try + { + var receipt = await operation.ConfigureAwait(false); + Complete(item, receipt, exception: null, canceled: false); + } + catch (OperationCanceledException exception) + { + Complete(item, receipt: null, exception, canceled: true); + } + catch (Exception exception) + { + Complete(item, receipt: null, exception, canceled: false); + } + } + + private void Complete( + WorkItem item, + GameActionReceipt? receipt, + Exception? exception, + bool canceled) + { + var stopped = false; + lock (_gate) + { + if (item.State != WorkState.Started) + { + return; + } + + item.State = WorkState.Completed; + _activeCount--; + stopped = !_accepting && _activeCount == 0; + } + + if (canceled) + { + item.Completion.TrySetCanceled(); + } + else if (exception is not null) + { + item.Completion.TrySetException(exception); + } + else + { + item.Completion.TrySetResult(receipt); + } + + if (stopped) + { + _stoppedCompletion.TrySetResult(null); + } + } + + private static async ValueTask AwaitRequiredReceiptAsync( + Task completion) + { + var receipt = await completion.ConfigureAwait(false); + return receipt ?? throw new InvalidOperationException("The queued execute handler returned a null receipt."); + } + + private enum WorkKind + { + Execute, + Recover, + } + + private enum WorkState + { + Queued, + Started, + Completed, + Canceled, + Stopped, + } + + private sealed class WorkItem + { + public WorkItem(QueuedGameActionHandler owner, GameActionIntent intent, WorkKind kind) + { + Owner = owner; + Intent = intent; + Kind = kind; + } + + public QueuedGameActionHandler Owner { get; } + + public GameActionIntent Intent { get; } + + public WorkKind Kind { get; } + + public TaskCompletionSource Completion { get; } = + new(TaskCreationOptions.RunContinuationsAsynchronously); + + public WorkState State { get; set; } + + public LinkedListNode? Node { get; set; } + + public CancellationTokenRegistration CancellationRegistration { get; set; } + } + + private sealed class CancellationState + { + public CancellationState(WorkItem item, CancellationToken token) + { + Item = item; + Token = token; + } + + public WorkItem Item { get; } + + public CancellationToken Token { get; } + } +} diff --git a/tests/OpenGameAgent.Tests/PublicApiCompatibilityTests.cs b/tests/OpenGameAgent.Tests/PublicApiCompatibilityTests.cs index 22a9a08..b802b63 100644 --- a/tests/OpenGameAgent.Tests/PublicApiCompatibilityTests.cs +++ b/tests/OpenGameAgent.Tests/PublicApiCompatibilityTests.cs @@ -5,7 +5,7 @@ namespace OpenGameAgent.Tests; public sealed class PublicApiCompatibilityTests { - private const string ApprovedApiHash = "436EAE0B0E6367F7532232DD799B22DB6CA3C2A01EC0E9D4009EE9B157427AEB"; + private const string ApprovedApiHash = "3C54A4325EAF055F6DEF45F051C70919901E61981CB2CD9EBB235B26F33E132C"; [Fact] public void RuntimePublicApiMatchesTheApprovedStableSurface() diff --git a/tests/OpenGameAgent.Tests/QueuedGameActionHandlerTests.cs b/tests/OpenGameAgent.Tests/QueuedGameActionHandlerTests.cs new file mode 100644 index 0000000..5c356d1 --- /dev/null +++ b/tests/OpenGameAgent.Tests/QueuedGameActionHandlerTests.cs @@ -0,0 +1,376 @@ +using System.Collections.Concurrent; +using Xunit; + +namespace OpenGameAgent.Tests; + +public sealed class QueuedGameActionHandlerTests +{ + [Fact] + public async Task PumpRunsExecuteAndRecoverOnItsBoundThreadInFifoOrder() + { + var calls = new List<(string OperationId, string Kind, int ThreadId, string? GenerationId)>(); + var inner = new CallbackHandler( + (intent, token) => + { + calls.Add((intent.OperationId, "execute", Environment.CurrentManagedThreadId, intent.GenerationId)); + Assert.False(token.CanBeCanceled); + return new ValueTask(GameActionReceipt.Committed(intent, "{}", 1)); + }, + (intent, token) => + { + calls.Add((intent.OperationId, "recover", Environment.CurrentManagedThreadId, intent.GenerationId)); + Assert.False(token.CanBeCanceled); + return new ValueTask(GameActionReceipt.Committed(intent, "{}", 1)); + }); + using var handler = new QueuedGameActionHandler(inner); + var pumpThread = Environment.CurrentManagedThreadId; + + var first = handler.ExecuteAsync(CreateIntent("operation-1"), CancellationToken.None).AsTask(); + var second = handler.RecoverAsync(CreateIntent("operation-2"), CancellationToken.None).AsTask(); + var third = handler.ExecuteAsync(CreateIntent("operation-3"), CancellationToken.None).AsTask(); + + Assert.False(first.IsCompleted); + Assert.Equal(2, handler.Pump(2)); + Assert.Equal(GameActionStatus.Committed, (await first).Status); + Assert.Equal(GameActionStatus.Committed, (await second)!.Status); + Assert.False(third.IsCompleted); + Assert.Equal(1, handler.Pump(2)); + Assert.Equal(GameActionStatus.Committed, (await third).Status); + Assert.Equal( + new[] + { + ("operation-1", "execute", pumpThread, (string?)"save-generation-1"), + ("operation-2", "recover", pumpThread, (string?)"save-generation-1"), + ("operation-3", "execute", pumpThread, (string?)"save-generation-1"), + }, + calls); + } + + [Fact] + public async Task QueuedCancellationRemovesWorkAndReleasesCapacity() + { + var executions = 0; + var inner = new CallbackHandler( + (intent, _) => + { + executions++; + return new ValueTask(GameActionReceipt.Committed(intent, "{}")); + }); + using var handler = new QueuedGameActionHandler(inner, maximumPendingActions: 1); + using var cancellation = new CancellationTokenSource(); + var canceled = handler.ExecuteAsync(CreateIntent("operation-canceled"), cancellation.Token).AsTask(); + + cancellation.Cancel(); + await Assert.ThrowsAnyAsync(async () => await canceled); + Assert.Equal(0, handler.PendingCount); + + var accepted = handler.ExecuteAsync(CreateIntent("operation-accepted"), CancellationToken.None).AsTask(); + Assert.Equal(1, handler.Pump(1)); + Assert.Equal(GameActionStatus.Committed, (await accepted).Status); + Assert.Equal(1, executions); + } + + [Fact] + public async Task CallerCancellationAfterStartDoesNotCancelWorldMutation() + { + var completion = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + CancellationToken receivedToken = default; + var intent = CreateIntent("operation-started"); + var inner = new CallbackHandler( + (_, token) => + { + receivedToken = token; + return new ValueTask(completion.Task); + }); + using var handler = new QueuedGameActionHandler(inner); + using var cancellation = new CancellationTokenSource(); + var action = handler.ExecuteAsync(intent, cancellation.Token).AsTask(); + + Assert.Equal(1, handler.Pump(1)); + Assert.Equal(1, handler.ActiveCount); + cancellation.Cancel(); + Assert.False(action.IsCompleted); + Assert.False(receivedToken.CanBeCanceled); + + completion.SetResult(GameActionReceipt.Committed(intent, "{}", 2)); + Assert.Equal(GameActionStatus.Committed, (await action).Status); + Assert.Equal(0, handler.ActiveCount); + } + + [Fact] + public async Task PendingAndActiveWorkAreIndependentlyBounded() + { + var completions = new ConcurrentDictionary>(); + var inner = new CallbackHandler( + (intent, _) => + { + var completion = new TaskCompletionSource( + TaskCreationOptions.RunContinuationsAsynchronously); + completions[intent.OperationId] = completion; + return new ValueTask(completion.Task); + }); + using var handler = new QueuedGameActionHandler( + inner, + maximumPendingActions: 2, + maximumActiveActions: 1); + + var first = handler.ExecuteAsync(CreateIntent("operation-1"), CancellationToken.None).AsTask(); + var second = handler.ExecuteAsync(CreateIntent("operation-2"), CancellationToken.None).AsTask(); + Assert.Throws(() => + handler.ExecuteAsync(CreateIntent("operation-overflow"), CancellationToken.None)); + + Assert.Equal(1, handler.Pump(2)); + Assert.Equal(1, handler.ActiveCount); + Assert.Equal(1, handler.PendingCount); + Assert.Equal(0, handler.Pump(2)); + + completions["operation-1"].SetResult( + GameActionReceipt.Committed(CreateIntent("operation-1"), "{}")); + Assert.True(SpinWait.SpinUntil(() => handler.ActiveCount == 0, TimeSpan.FromSeconds(5))); + Assert.Equal(1, handler.Pump(2)); + completions["operation-2"].SetResult( + GameActionReceipt.Committed(CreateIntent("operation-2"), "{}")); + Assert.Equal(GameActionStatus.Committed, (await first).Status); + Assert.Equal(GameActionStatus.Committed, (await second).Status); + } + + [Fact] + public async Task HandlerFailureDoesNotBlockTheFollowingQueueItem() + { + var inner = new CallbackHandler( + (intent, _) => + { + if (intent.OperationId == "operation-failed") + { + throw new InvalidOperationException("failed safely"); + } + + return new ValueTask(GameActionReceipt.Committed(intent, "{}")); + }); + using var handler = new QueuedGameActionHandler(inner); + var failed = handler.ExecuteAsync(CreateIntent("operation-failed"), CancellationToken.None).AsTask(); + var succeeded = handler.ExecuteAsync(CreateIntent("operation-succeeded"), CancellationToken.None).AsTask(); + + Assert.Equal(2, handler.Pump(2)); + var exception = await Assert.ThrowsAsync(async () => await failed); + Assert.Equal("failed safely", exception.Message); + Assert.Equal(GameActionStatus.Committed, (await succeeded).Status); + } + + [Fact] + public async Task ConcurrentProducersCannotExceedPendingCapacity() + { + const int capacity = 32; + var inner = new CallbackHandler( + (intent, _) => new ValueTask(GameActionReceipt.Committed(intent, "{}"))); + using var handler = new QueuedGameActionHandler(inner, maximumPendingActions: capacity); + + var submissions = await Task.WhenAll( + Enumerable.Range(0, capacity * 4) + .Select(index => Task.Run(() => + { + try + { + return ( + Completion: (Task?)handler.ExecuteAsync( + CreateIntent($"operation-concurrent-{index}"), + CancellationToken.None).AsTask(), + Accepted: true); + } + catch (GameRuntimeLimitException) + { + return (Completion: (Task?)null, Accepted: false); + } + }))); + + var accepted = submissions + .Where(static submission => submission.Accepted) + .Select(static submission => submission.Completion!) + .ToArray(); + Assert.Equal(capacity, accepted.Length); + Assert.Equal(capacity, handler.PendingCount); + Assert.Equal(capacity, handler.Pump(capacity)); + Assert.All(await Task.WhenAll(accepted), receipt => Assert.Equal(GameActionStatus.Committed, receipt.Status)); + } + + [Fact] + public async Task PumpCannotBeReenteredByAnActionHandler() + { + QueuedGameActionHandler? handler = null; + Exception? reentrantFailure = null; + var inner = new CallbackHandler( + (intent, _) => + { + reentrantFailure = Record.Exception(() => handler!.Pump(1)); + return new ValueTask(GameActionReceipt.Committed(intent, "{}")); + }); + handler = new QueuedGameActionHandler(inner); + using (handler) + { + var action = handler.ExecuteAsync(CreateIntent("operation-reentrant"), CancellationToken.None).AsTask(); + Assert.Equal(1, handler.Pump(1)); + Assert.Equal(GameActionStatus.Committed, (await action).Status); + Assert.IsType(reentrantFailure); + } + } + + [Fact] + public async Task AsyncDisposeRejectsPendingWorkAndWaitsForStartedWork() + { + var intent = CreateIntent("operation-active"); + var completion = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var inner = new CallbackHandler( + (current, _) => current.OperationId == intent.OperationId + ? new ValueTask(completion.Task) + : new ValueTask(GameActionReceipt.Committed(current, "{}"))); + var handler = new QueuedGameActionHandler(inner); + var active = handler.ExecuteAsync(intent, CancellationToken.None).AsTask(); + var pending = handler.ExecuteAsync(CreateIntent("operation-pending"), CancellationToken.None).AsTask(); + Assert.Equal(1, handler.Pump(1)); + + var disposing = handler.DisposeAsync().AsTask(); + await Assert.ThrowsAsync(async () => await pending); + Assert.Throws(() => + handler.ExecuteAsync(CreateIntent("operation-rejected"), CancellationToken.None)); + Assert.False(disposing.IsCompleted); + Assert.False(handler.IsAccepting); + + completion.SetResult(GameActionReceipt.Committed(intent, "{}")); + Assert.Equal(GameActionStatus.Committed, (await active).Status); + await disposing; + Assert.Equal(0, handler.ActiveCount); + } + + [Fact] + public void PumpRejectsASecondThreadAfterBinding() + { + using var handler = new QueuedGameActionHandler(new CallbackHandler()); + Assert.Equal(0, handler.Pump(1)); + + Exception? exception = null; + var secondThread = new Thread(() => exception = Record.Exception(() => handler.Pump(1))); + secondThread.Start(); + Assert.True(secondThread.Join(TimeSpan.FromSeconds(5))); + Assert.IsType(exception); + } + + [Fact] + public async Task DurableRestartRecoversDispatchedWorkWithoutReExecutingIt() + { + var journal = new InMemoryGameActionJournal(); + var firstInner = new CallbackHandler(); + var firstQueue = new QueuedGameActionHandler(firstInner); + var firstDispatcher = new DurableGameActionDispatcher(journal, firstQueue); + var intent = CreateIntent("operation-restart"); + var firstAttempt = firstDispatcher.ExecuteAsync(intent, CancellationToken.None).AsTask(); + await WaitUntilAsync(() => firstQueue.PendingCount == 1); + + firstQueue.Stop(); + var uncertain = await firstAttempt; + Assert.Equal(GameActionStatus.Uncertain, uncertain.Status); + Assert.Equal(0, firstInner.ExecuteCount); + + var secondInner = new CallbackHandler( + execute: null, + recover: (current, _) => + new ValueTask(GameActionReceipt.Committed(current, "{}", 3))); + using var secondQueue = new QueuedGameActionHandler(secondInner); + var restartedDispatcher = new DurableGameActionDispatcher(journal, secondQueue); + var recovery = restartedDispatcher.ReconcileAsync(intent.OperationId, CancellationToken.None).AsTask(); + await WaitUntilAsync(() => secondQueue.PendingCount == 1); + + Assert.Equal(1, secondQueue.Pump(1)); + Assert.Equal(GameActionStatus.Committed, (await recovery).Status); + Assert.Equal(0, secondInner.ExecuteCount); + Assert.Equal(1, secondInner.RecoverCount); + } + + [Fact] + public async Task CanceledDurableQueueItemIsRecoveredInsteadOfBlindlyExecuted() + { + var journal = new InMemoryGameActionJournal(); + var inner = new CallbackHandler( + execute: null, + recover: (intent, _) => + new ValueTask(GameActionReceipt.Committed(intent, "{}", 4))); + using var handler = new QueuedGameActionHandler(inner); + var dispatcher = new DurableGameActionDispatcher(journal, handler); + var intent = CreateIntent("operation-cancel-recover"); + using var cancellation = new CancellationTokenSource(); + var firstAttempt = dispatcher.ExecuteAsync(intent, cancellation.Token).AsTask(); + await WaitUntilAsync(() => handler.PendingCount == 1); + + cancellation.Cancel(); + await Assert.ThrowsAnyAsync(async () => await firstAttempt); + Assert.Equal(0, inner.ExecuteCount); + Assert.Equal(0, handler.PendingCount); + + var recovery = dispatcher.ReconcileAsync(intent.OperationId, CancellationToken.None).AsTask(); + await WaitUntilAsync(() => handler.PendingCount == 1); + Assert.Equal(1, handler.Pump(1)); + Assert.Equal(GameActionStatus.Committed, (await recovery).Status); + Assert.Equal(0, inner.ExecuteCount); + Assert.Equal(1, inner.RecoverCount); + } + + private static GameActionIntent CreateIntent(string operationId) => + new( + operationId, + "input-1", + "session-1", + "actor-1", + "move", + "{}", + new GameMoment("timeline-1", 10), + expectedRevision: 1, + generationId: "save-generation-1"); + + private static async Task WaitUntilAsync(Func predicate) + { + using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5)); + while (!predicate()) + { + await Task.Delay(10, timeout.Token); + } + } + + private sealed class CallbackHandler : IGameActionHandler + { + private readonly Func>? _execute; + private readonly Func>? _recover; + private int _executeCount; + private int _recoverCount; + + public CallbackHandler( + Func>? execute = null, + Func>? recover = null) + { + _execute = execute; + _recover = recover; + } + + public int ExecuteCount => Volatile.Read(ref _executeCount); + + public int RecoverCount => Volatile.Read(ref _recoverCount); + + public ValueTask ExecuteAsync( + GameActionIntent intent, + CancellationToken cancellationToken) + { + Interlocked.Increment(ref _executeCount); + return _execute is null + ? throw new InvalidOperationException("Execute was not expected.") + : _execute(intent, cancellationToken); + } + + public ValueTask RecoverAsync( + GameActionIntent intent, + CancellationToken cancellationToken) + { + Interlocked.Increment(ref _recoverCount); + return _recover is null + ? new ValueTask((GameActionReceipt?)null) + : _recover(intent, cancellationToken); + } + } +}