From c7ee2d4e33df0e61ccef84f0de1b8cdf2603104e Mon Sep 17 00:00:00 2001 From: jeremydixon22 Date: Mon, 10 Aug 2026 06:26:23 -0400 Subject: [PATCH] test: make durable wait races deterministic --- PUBLIC-EXPORT-MANIFEST.json | 6 +-- ...ExecutionRuntimeAdapterConformanceTests.cs | 40 ++++++++++++++----- .../ExecutionRuntimeTests.cs | 22 +++++++++- 3 files changed, 53 insertions(+), 15 deletions(-) diff --git a/PUBLIC-EXPORT-MANIFEST.json b/PUBLIC-EXPORT-MANIFEST.json index d73d872..296b5f2 100644 --- a/PUBLIC-EXPORT-MANIFEST.json +++ b/PUBLIC-EXPORT-MANIFEST.json @@ -3339,7 +3339,7 @@ { "mode": "644", "path": "tests/Vyral.Tests.Aws/AwsDynamoExecutionRuntimeAdapterConformanceTests.cs", - "sha256": "cc5dc18c4c0ea9e43b21f9ef66c95bbbf1cebc7dd2bf1f0c561c2ab8b006b18d" + "sha256": "e4fe9df308809d38707710e2dccbfddca6976f55c87f6f40c8ba9df3130faf48" }, { "mode": "644", @@ -3739,7 +3739,7 @@ { "mode": "644", "path": "tests/Vyral.Tests.Local/ExecutionRuntimeTests.cs", - "sha256": "da20289a6d5284580a941e7bee380cabdf8a76f43922f3de398fa8aec1964009" + "sha256": "ed6e31821e875fd2ac866f35b75dd5ab538fc20d63e60d7e153cd2536ddea258" }, { "mode": "644", @@ -4079,5 +4079,5 @@ ], "schemaVersion": 1, "sourceDirty": false, - "treeSha256": "409f31c6b29eb940bf9076e77e68fa715ac941d41437f4dec9b3d19eb51810e1" + "treeSha256": "671ee8905302737f33c2abf43f100cf13f58a64ec49afe3f2bfa60986d29e020" } diff --git a/tests/Vyral.Tests.Aws/AwsDynamoExecutionRuntimeAdapterConformanceTests.cs b/tests/Vyral.Tests.Aws/AwsDynamoExecutionRuntimeAdapterConformanceTests.cs index 6f0c38e..8e768e1 100644 --- a/tests/Vyral.Tests.Aws/AwsDynamoExecutionRuntimeAdapterConformanceTests.cs +++ b/tests/Vyral.Tests.Aws/AwsDynamoExecutionRuntimeAdapterConformanceTests.cs @@ -24,6 +24,12 @@ public async Task AwsExecutionRuntime_MatchesPublishedQualificationProfile() } protected override Task CreateExternalWorkerRuntimeAsync() + { + var fixture = CreateExternalWorkerRuntimeFixture(); + return Task.FromResult(fixture.Runtime); + } + + private static (ExternalExecutionWorkerRuntimeFixture Runtime, InMemoryAwsDynamoExecutionStateStore State) CreateExternalWorkerRuntimeFixture() { var handler = new ExecutionHandlerDescriptor { @@ -32,8 +38,9 @@ protected override Task CreateExternalWor DisplayName = "AWS external worker conformance handler" }; var dispatcher = new CapturingExecutionRunDispatcher(); + var state = new InMemoryAwsDynamoExecutionStateStore(); var runtime = new AwsDynamoExecutionRuntimeAdapter( - new InMemoryAwsDynamoExecutionStateStore(), + state, dispatcher, new AwsDynamoExecutionRuntimeOptions { @@ -46,12 +53,14 @@ protected override Task CreateExternalWor } ] }); - return Task.FromResult(new ExternalExecutionWorkerRuntimeFixture - { - Adapter = runtime, - Worker = runtime, - Handler = handler - }); + return ( + new ExternalExecutionWorkerRuntimeFixture + { + Adapter = runtime, + Worker = runtime, + Handler = handler + }, + state); } [Fact] @@ -503,7 +512,8 @@ public async Task AwsRuntime_CancellationFencePreventsAStaleWorkerCompletion() [Fact] public async Task AwsRuntime_EventAndTimeoutRaceConsumesExactlyOneDurableWait() { - var fixture = await CreateExternalWorkerRuntimeAsync(); + var created = CreateExternalWorkerRuntimeFixture(); + var fixture = created.Runtime; fixture.Worker.RegisterExternalHandler(fixture.Handler); var accepted = await fixture.Adapter.StartRunAsync(new ExecutionRunRequest { HandlerId = fixture.Handler.HandlerId }); var firstLease = Assert.IsType(await fixture.Worker.LeaseNextRunAsync(new ExecutionExternalWorkerLeaseRequest @@ -519,11 +529,11 @@ public async Task AwsRuntime_EventAndTimeoutRaceConsumesExactlyOneDurableWait() WorkerId = firstLease.WorkerId, Kind = ExecutionExternalWorkerWaitKinds.ExternalEvent, Name = "approval", - TimeoutAtUtc = DateTime.UtcNow.AddMilliseconds(20) + TimeoutAtUtc = DateTime.UtcNow.AddHours(1) }); Assert.True(suspended.Suspended); - await Task.Delay(40); + created.State.MakeWaitDue(accepted.Id); var eventTask = fixture.Adapter.RaiseEventAsync(new ExecutionExternalEventRequest { RunId = accepted.Id, @@ -633,6 +643,16 @@ private sealed class InMemoryAwsDynamoExecutionStateStore : IAwsDynamoExecutionS private readonly Dictionary _waitOutcomes = new(StringComparer.Ordinal); private readonly HashSet _consumedEvents = new(StringComparer.Ordinal); + public void MakeWaitDue(string runId) + { + lock (_gate) + { + if (!_waits.TryGetValue(runId, out var wait)) + throw new InvalidOperationException($"Execution wait for run '{runId}' was not found."); + wait.FireAtUtc = DateTime.UtcNow.AddSeconds(-1); + } + } + public Task CreateRunAsync(ExecutionRun run, CancellationToken ct = default) { lock (_gate) diff --git a/tests/Vyral.Tests.Local/ExecutionRuntimeTests.cs b/tests/Vyral.Tests.Local/ExecutionRuntimeTests.cs index 74649d8..caff307 100644 --- a/tests/Vyral.Tests.Local/ExecutionRuntimeTests.cs +++ b/tests/Vyral.Tests.Local/ExecutionRuntimeTests.cs @@ -1100,8 +1100,8 @@ public async Task LocalRuntime_DurablyWaitsForTimerAndTimeoutWithoutHoldingWorke var timeoutRun = await runtime.StartRunAsync(new ExecutionRunRequest { HandlerId = TimeoutWaitHandler.HandlerId }); await timerHandler.WaitRegistered.Task.WaitAsync(TimeSpan.FromSeconds(5)); await timeoutHandler.WaitRegistered.Task.WaitAsync(TimeSpan.FromSeconds(5)); - await WaitForRunAsync(runtime, timerRun.Id, ExecutionRunStatuses.Waiting); - await WaitForRunAsync(runtime, timeoutRun.Id, ExecutionRunStatuses.Waiting); + await WaitForHistoryEventAsync(runtime, timerRun.Id, ExecutionEventTypes.WaitRegistered); + await WaitForHistoryEventAsync(runtime, timeoutRun.Id, ExecutionEventTypes.WaitRegistered); var timerCompleted = await WaitForRunAsync(runtime, timerRun.Id, ExecutionRunStatuses.Succeeded); var timeoutCompleted = await WaitForRunAsync(runtime, timeoutRun.Id, ExecutionRunStatuses.Succeeded); @@ -1296,6 +1296,24 @@ private static async Task WaitForRunAsync(IExecutionRuntime runtim throw new InvalidOperationException($"Run {id} did not reach {status}. Last status: {run?.Status ?? "(missing)"}. History: {historySummary}"); } + private static async Task WaitForHistoryEventAsync(IExecutionRuntime runtime, string id, string eventType) + { + IReadOnlyList history = Array.Empty(); + for (var i = 0; i < 200; i++) + { + history = await runtime.GetHistoryAsync(id); + if (history.Any(item => item.Type == eventType)) + { + return; + } + + await Task.Delay(25); + } + + var historySummary = string.Join(", ", history.Select(item => $"{item.Type}:{item.Status}")); + throw new InvalidOperationException($"Run {id} did not record {eventType}. History: {historySummary}"); + } + private static async Task WaitForMissingRunAsync(IExecutionRuntime runtime, string id) { ExecutionRun? run = null;