test: make durable wait races deterministic

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"
 }

tests/Vyral.Tests.Aws/AwsDynamoExecutionRuntimeAdapterConformanceTests.cs

@@ -24,6 +24,12 @@ public async Task AwsExecutionRuntime_MatchesPublishedQualificationProfile()
     }
 
     protected override Task<ExternalExecutionWorkerRuntimeFixture> 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<ExternalExecutionWorkerRuntimeFixture> 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<ExternalExecutionWorkerRuntimeFixture> 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<ExecutionExternalWorkerLease>(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<string, ExecutionWaitResult> _waitOutcomes = new(StringComparer.Ordinal);
         private readonly HashSet<string> _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)

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<ExecutionRun> 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<ExecutionTraceEvent> history = Array.Empty<ExecutionTraceEvent>();
+        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;