| | | 1 | | using System.Collections.Concurrent; |
| | | 2 | | using System.Diagnostics.CodeAnalysis; |
| | | 3 | | |
| | | 4 | | namespace AsyncResponse.Testing; |
| | | 5 | | |
| | | 6 | | /// <summary> |
| | | 7 | | /// Durable-flow test harness: runs real flows on the in-memory engine |
| | | 8 | | /// (<see cref="AsyncResponseTestHarness"/>) with step-level observation, scripted replies to |
| | | 9 | | /// awaited steps, deterministic crash injection at step boundaries, and virtual-time control for |
| | | 10 | | /// durable timers — without touching the flow class under test. |
| | | 11 | | /// <para> |
| | | 12 | | /// A flow can be driven two ways. <see cref="FlowRunHandle.ExecuteDirectAsync"/> invokes the |
| | | 13 | | /// executor inline — single-threaded and fully deterministic, ideal for crash-at-checkpoint |
| | | 14 | | /// matrices (an injected crash surfaces as the returned attempt's |
| | | 15 | | /// <see cref="SimulatedCrashException"/>; call it again to "restart"). Alternatively, starting via |
| | | 16 | | /// <see cref="StartFlowAsync{TFlow, TInput}"/> runs the whole production pipeline: the worker |
| | | 17 | | /// queue executes the run, crashes ride the transport's redelivery-with-backoff, and |
| | | 18 | | /// <see cref="AsyncResponseTestHarness.AdvanceAsync"/> drives retries, timers, and schedules. |
| | | 19 | | /// </para> |
| | | 20 | | /// </summary> |
| | | 21 | | public sealed class FlowTestHarness : IAsyncDisposable |
| | | 22 | | { |
| | | 23 | | private readonly FlowProbe _probe; |
| | | 24 | | |
| | 134 | 25 | | private FlowTestHarness(AsyncResponseTestHarness engine, FlowProbe probe) |
| | | 26 | | { |
| | 134 | 27 | | Engine = engine; |
| | 134 | 28 | | _probe = probe; |
| | 134 | 29 | | } |
| | | 30 | | |
| | | 31 | | /// <summary>Builds the engine (with the flow probe installed) and starts it.</summary> |
| | | 32 | | public static async Task<FlowTestHarness> StartAsync(Action<AsyncResponseTestHarnessOptions>? configure = null) |
| | | 33 | | { |
| | 134 | 34 | | var probe = new FlowProbe(); |
| | 134 | 35 | | var engine = await AsyncResponseTestHarness.StartAsync(options => |
| | 134 | 36 | | { |
| | 134 | 37 | | configure?.Invoke(options); |
| | 134 | 38 | | options.FlowObservers.Add(probe); |
| | 268 | 39 | | }).ConfigureAwait(false); |
| | 134 | 40 | | return new FlowTestHarness(engine, probe); |
| | 134 | 41 | | } |
| | | 42 | | |
| | | 43 | | /// <summary>The underlying engine harness (publisher, builder, clock, restart).</summary> |
| | 2337 | 44 | | public AsyncResponseTestHarness Engine { get; } |
| | | 45 | | |
| | | 46 | | /// <summary>The virtual clock (shorthand for <c>Engine.Clock</c>).</summary> |
| | 24 | 47 | | public VirtualTimeProvider Clock => Engine.Clock; |
| | | 48 | | |
| | | 49 | | /// <summary>Advances virtual time (shorthand for <c>Engine.AdvanceAsync</c>).</summary> |
| | 842 | 50 | | public Task AdvanceAsync(TimeSpan delta) => Engine.AdvanceAsync(delta); |
| | | 51 | | |
| | | 52 | | /// <summary> |
| | | 53 | | /// Starts a flow through the production pipeline (ledger create + worker-queue wake-up) and |
| | | 54 | | /// returns its handle. |
| | | 55 | | /// </summary> |
| | | 56 | | public async Task<FlowRunHandle> StartFlowAsync<[DynamicallyAccessedMembers(DynamicallyAccessedMemberTypes.PublicCon |
| | | 57 | | TInput input, |
| | | 58 | | string? flowId = null) |
| | | 59 | | where TFlow : class, IDurableFlow<TInput> |
| | | 60 | | { |
| | 124 | 61 | | var id = await Engine.Flows.StartAsync<TFlow, TInput>(input, flowId).ConfigureAwait(false); |
| | 102 | 62 | | return Attach(id); |
| | 102 | 63 | | } |
| | | 64 | | |
| | | 65 | | /// <summary>Attaches a handle to an existing run (e.g. one started by a cron schedule).</summary> |
| | | 66 | | public FlowRunHandle Attach(string flowId) |
| | 114 | 67 | | => new(this, flowId); |
| | | 68 | | |
| | | 69 | | /// <summary> |
| | | 70 | | /// Arms a one-shot crash that fires when <paramref name="stepName"/> is next about to execute |
| | | 71 | | /// (before any of its side effects). The execution attempt fails with |
| | | 72 | | /// <see cref="SimulatedCrashException"/> and the run resumes from its last checkpoint on the |
| | | 73 | | /// next delivery — the flow class under test needs no instrumentation. Pass |
| | | 74 | | /// <paramref name="flowId"/> when more than one run (concurrent flows, or a parent and its |
| | | 75 | | /// child) can reach the step: an unscoped crash fires on whichever run gets there first, so a |
| | | 76 | | /// test could pass without the intended run ever exercising its recovery path. |
| | | 77 | | /// </summary> |
| | | 78 | | public void CrashBeforeStep(string stepName, string? flowId = null) |
| | 16 | 79 | | => _probe.ArmCrash(stepName, beforeStep: true, flowId); |
| | | 80 | | |
| | | 81 | | /// <summary> |
| | | 82 | | /// Arms a one-shot crash that fires right after <paramref name="stepName"/>'s completion |
| | | 83 | | /// checkpoint persists — the classic "died between the checkpoint and the next step" window. |
| | | 84 | | /// Pass <paramref name="flowId"/> to pin the crash to one run; see |
| | | 85 | | /// <see cref="CrashBeforeStep"/>. |
| | | 86 | | /// </summary> |
| | | 87 | | public void CrashAfterStep(string stepName, string? flowId = null) |
| | 12 | 88 | | => _probe.ArmCrash(stepName, beforeStep: false, flowId); |
| | | 89 | | |
| | 1012 | 90 | | internal FlowProbe Probe => _probe; |
| | | 91 | | |
| | | 92 | | /// <inheritdoc/> |
| | 134 | 93 | | public ValueTask DisposeAsync() => Engine.DisposeAsync(); |
| | | 94 | | } |
| | | 95 | | |
| | | 96 | | /// <summary>A handle on one flow run under test.</summary> |
| | | 97 | | public sealed class FlowRunHandle |
| | | 98 | | { |
| | | 99 | | private readonly FlowTestHarness _harness; |
| | | 100 | | |
| | | 101 | | internal FlowRunHandle(FlowTestHarness harness, string flowId) |
| | | 102 | | { |
| | | 103 | | _harness = harness; |
| | | 104 | | FlowId = flowId; |
| | | 105 | | } |
| | | 106 | | |
| | | 107 | | /// <summary>The run id.</summary> |
| | | 108 | | public string FlowId { get; } |
| | | 109 | | |
| | | 110 | | /// <summary>Loads the run's current persisted state (its ledger), or <c>null</c> when none exists.</summary> |
| | | 111 | | public Task<FlowState?> GetStateAsync() |
| | | 112 | | => _harness.Engine.Flows.GetStateAsync(FlowId); |
| | | 113 | | |
| | | 114 | | /// <summary> |
| | | 115 | | /// Executes the run inline on the calling thread (no worker queue): returns when the executor |
| | | 116 | | /// attempt finishes — completed, suspended (timer/child parked), or failed. An injected crash |
| | | 117 | | /// or a step exception propagates to the caller; call again to simulate the next delivery. |
| | | 118 | | /// </summary> |
| | | 119 | | public async Task ExecuteDirectAsync() |
| | | 120 | | { |
| | | 121 | | // Counted so AdvanceAsync's settle treats this attempt like a worker job: it holds no |
| | | 122 | | // worker slot, so a step it parks would otherwise tip ParkedCount past OutstandingJobs |
| | | 123 | | // and let the clock advance mid-attempt. |
| | | 124 | | _harness.Engine.OnDirectRunStarted(); |
| | | 125 | | try |
| | | 126 | | { |
| | | 127 | | await _harness.Engine.FlowExecutor.ExecuteAsync(FlowId).ConfigureAwait(false); |
| | | 128 | | } |
| | | 129 | | finally |
| | | 130 | | { |
| | | 131 | | _harness.Engine.OnDirectRunFinished(); |
| | | 132 | | } |
| | | 133 | | } |
| | | 134 | | |
| | | 135 | | /// <summary>Re-enqueues the run on the worker queue (the operator's resume action).</summary> |
| | | 136 | | public Task ResumeAsync() |
| | | 137 | | => _harness.Engine.Flows.ResumeAsync(FlowId); |
| | | 138 | | |
| | | 139 | | /// <summary>How many times <paramref name="stepName"/> actually started executing (memoized skips excluded).</summa |
| | | 140 | | public int StepExecutions(string stepName) |
| | | 141 | | => _harness.Probe.CountEvents(FlowId, stepName, FlowProbe.EventKind.Starting); |
| | | 142 | | |
| | | 143 | | /// <summary>Every recorded step event for this run, in order.</summary> |
| | | 144 | | public IReadOnlyList<FlowProbeEvent> Events |
| | | 145 | | => _harness.Probe.EventsFor(FlowId); |
| | | 146 | | |
| | | 147 | | /// <summary> |
| | | 148 | | /// Waits (bounded by the harness real-time guard) until <paramref name="stepName"/> is parked |
| | | 149 | | /// awaiting its response, and returns the correlation id to answer. |
| | | 150 | | /// </summary> |
| | | 151 | | public async Task<string> WaitForAwaitingStepAsync(string stepName) |
| | | 152 | | { |
| | | 153 | | var stepEvent = await _harness.Probe.WaitForAsync( |
| | | 154 | | FlowId, |
| | | 155 | | e => e.Kind == FlowProbe.EventKind.Waiting && e.Step.Kind == DurableFlowStepKind.Awaited && e.Step.StepName |
| | | 156 | | _harness.Engine.RealTimeGuard, |
| | | 157 | | $"step '{stepName}' of flow '{FlowId}' to be awaiting a response").ConfigureAwait(false); |
| | | 158 | | return stepEvent.Step.CorrelationId!; |
| | | 159 | | } |
| | | 160 | | |
| | | 161 | | /// <summary>Waits until a timer step is parked, returning its due time (advance the clock past it to wake the run). |
| | | 162 | | public async Task<DateTime> WaitForTimerStepAsync(string stepName) |
| | | 163 | | { |
| | | 164 | | var stepEvent = await _harness.Probe.WaitForAsync( |
| | | 165 | | FlowId, |
| | | 166 | | e => e.Kind == FlowProbe.EventKind.Waiting && e.Step.Kind == DurableFlowStepKind.Timer && e.Step.StepName == |
| | | 167 | | _harness.Engine.RealTimeGuard, |
| | | 168 | | $"timer step '{stepName}' of flow '{FlowId}' to be sleeping").ConfigureAwait(false); |
| | | 169 | | return stepEvent.Step.WakeAtUtc!.Value; |
| | | 170 | | } |
| | | 171 | | |
| | | 172 | | /// <summary>Waits until the given step's completion checkpoint persists.</summary> |
| | | 173 | | public Task WaitForStepCompletedAsync(string stepName) |
| | | 174 | | => _harness.Probe.WaitForAsync( |
| | | 175 | | FlowId, |
| | | 176 | | e => e.Kind == FlowProbe.EventKind.Completed && e.Step.StepName == stepName, |
| | | 177 | | _harness.Engine.RealTimeGuard, |
| | | 178 | | $"step '{stepName}' of flow '{FlowId}' to complete"); |
| | | 179 | | |
| | | 180 | | /// <summary>Waits until the run reaches a terminal status and returns it.</summary> |
| | | 181 | | public async Task<FlowRunStatus> WaitForFinishedAsync() |
| | | 182 | | { |
| | | 183 | | var finished = await _harness.Probe.WaitForRunAsync( |
| | | 184 | | FlowId, |
| | | 185 | | _harness.Engine.RealTimeGuard, |
| | | 186 | | $"flow '{FlowId}' to finish").ConfigureAwait(false); |
| | | 187 | | return finished.Status; |
| | | 188 | | } |
| | | 189 | | |
| | | 190 | | /// <summary> |
| | | 191 | | /// Answers the run's currently awaited step: replies to the correlation id of the most recent |
| | | 192 | | /// awaited-step wait (optionally the named step's). Progress-aware steps take several replies — |
| | | 193 | | /// non-terminal payloads keep the wait open exactly as in production. |
| | | 194 | | /// </summary> |
| | | 195 | | public async Task ReplyAsync<T>(T response, string? stepName = null) where T : IAsyncResponsePayload |
| | | 196 | | { |
| | | 197 | | var correlationId = await NextReplyTargetAsync(stepName).ConfigureAwait(false); |
| | | 198 | | await _harness.Engine.PublishAsync(response, correlationId).ConfigureAwait(false); |
| | | 199 | | } |
| | | 200 | | |
| | | 201 | | /// <summary>Fails the run's currently awaited step with an exception (a remote failure).</summary> |
| | | 202 | | public async Task ReplyExceptionAsync(Exception exception, string? stepName = null) |
| | | 203 | | { |
| | | 204 | | var correlationId = await NextReplyTargetAsync(stepName).ConfigureAwait(false); |
| | | 205 | | await _harness.Engine.PublishExceptionAsync(exception, correlationId).ConfigureAwait(false); |
| | | 206 | | } |
| | | 207 | | |
| | | 208 | | // One call resolves the live un-answered wait or parks for the NEXT one — replaying history |
| | | 209 | | // here (as the old fallback did) handed back an already-answered or abandoned correlation id, |
| | | 210 | | // and the reply published to it was silently dropped. |
| | | 211 | | private Task<string> NextReplyTargetAsync(string? stepName) |
| | | 212 | | => _harness.Probe.WaitForNextAwaitedAsync( |
| | | 213 | | FlowId, |
| | | 214 | | stepName, |
| | | 215 | | _harness.Engine.RealTimeGuard, |
| | | 216 | | stepName is null |
| | | 217 | | ? $"flow '{FlowId}' to be awaiting any step" |
| | | 218 | | : $"step '{stepName}' of flow '{FlowId}' to be awaiting a response"); |
| | | 219 | | } |
| | | 220 | | |
| | | 221 | | /// <summary>One observation recorded by the flow probe.</summary> |
| | | 222 | | /// <param name="Kind">Which lifecycle point.</param> |
| | | 223 | | /// <param name="Step">The step event as reported by the executor.</param> |
| | | 224 | | public readonly record struct FlowProbeEvent(FlowProbe.EventKind Kind, DurableFlowStepEvent Step); |
| | | 225 | | |
| | | 226 | | /// <summary> |
| | | 227 | | /// The harness's <see cref="IDurableFlowExecutionObserver"/>: records step events per run, wakes |
| | | 228 | | /// waiters, and throws armed one-shot crashes. Public only for the event-kind enum and the |
| | | 229 | | /// recorded-event type; tests interact through <see cref="FlowTestHarness"/>. |
| | | 230 | | /// </summary> |
| | | 231 | | public sealed class FlowProbe : IDurableFlowExecutionObserver |
| | | 232 | | { |
| | | 233 | | /// <summary>Step lifecycle points the probe records.</summary> |
| | | 234 | | public enum EventKind |
| | | 235 | | { |
| | | 236 | | /// <summary>The step is about to execute (<see cref="IDurableFlowExecutionObserver.OnStepStartingAsync"/>).</su |
| | | 237 | | Starting = 0, |
| | | 238 | | |
| | | 239 | | /// <summary>The step is durably parked (<see cref="IDurableFlowExecutionObserver.OnStepWaitingAsync"/>).</summa |
| | | 240 | | Waiting = 1, |
| | | 241 | | |
| | | 242 | | /// <summary>The step's checkpoint persisted (<see cref="IDurableFlowExecutionObserver.OnStepCompletedAsync"/>). |
| | | 243 | | Completed = 2 |
| | | 244 | | } |
| | | 245 | | |
| | | 246 | | private readonly object _gate = new(); |
| | | 247 | | private readonly ConcurrentDictionary<string, List<FlowProbeEvent>> _events = new(StringComparer.Ordinal); |
| | | 248 | | private readonly ConcurrentDictionary<string, DurableFlowRunEvent> _finished = new(StringComparer.Ordinal); |
| | | 249 | | private readonly List<Waiter> _waiters = []; |
| | | 250 | | private readonly List<RunWaiter> _runWaiters = []; |
| | | 251 | | private (string Step, bool Before, string? FlowId)? _armedCrash; |
| | | 252 | | |
| | | 253 | | internal void ArmCrash(string stepName, bool beforeStep, string? flowId = null) |
| | | 254 | | { |
| | | 255 | | lock (_gate) |
| | | 256 | | { |
| | | 257 | | // One slot, so a second arm would silently discard a still-unfired first one — and |
| | | 258 | | // the test would pass without ever exercising the crash it asked for. Fail loudly |
| | | 259 | | // instead; arm the next crash after the current one fires. |
| | | 260 | | if (_armedCrash is { } armed) |
| | | 261 | | throw new InvalidOperationException( |
| | | 262 | | $"A crash is already armed for step '{armed.Step}' ({(armed.Before ? "before" : "after")} the step) |
| | | 263 | | "arm one crash at a time, after the previous one has fired."); |
| | | 264 | | |
| | | 265 | | _armedCrash = (stepName, beforeStep, flowId); |
| | | 266 | | } |
| | | 267 | | } |
| | | 268 | | |
| | | 269 | | ValueTask IDurableFlowExecutionObserver.OnStepStartingAsync(DurableFlowStepEvent step) |
| | | 270 | | { |
| | | 271 | | Record(EventKind.Starting, step); |
| | | 272 | | MaybeCrash(step.StepName, beforeStep: true, step.FlowId); |
| | | 273 | | return default; |
| | | 274 | | } |
| | | 275 | | |
| | | 276 | | ValueTask IDurableFlowExecutionObserver.OnStepWaitingAsync(DurableFlowStepEvent step) |
| | | 277 | | { |
| | | 278 | | Record(EventKind.Waiting, step); |
| | | 279 | | return default; |
| | | 280 | | } |
| | | 281 | | |
| | | 282 | | ValueTask IDurableFlowExecutionObserver.OnStepCompletedAsync(DurableFlowStepEvent step) |
| | | 283 | | { |
| | | 284 | | Record(EventKind.Completed, step); |
| | | 285 | | MaybeCrash(step.StepName, beforeStep: false, step.FlowId); |
| | | 286 | | return default; |
| | | 287 | | } |
| | | 288 | | |
| | | 289 | | ValueTask IDurableFlowExecutionObserver.OnRunFinishedAsync(DurableFlowRunEvent run) |
| | | 290 | | { |
| | | 291 | | _finished[run.FlowId] = run; |
| | | 292 | | RunWaiter[] due; |
| | | 293 | | lock (_gate) |
| | | 294 | | { |
| | | 295 | | due = [.. _runWaiters.Where(w => w.FlowId == run.FlowId)]; |
| | | 296 | | _runWaiters.RemoveAll(w => w.FlowId == run.FlowId); |
| | | 297 | | } |
| | | 298 | | |
| | | 299 | | foreach (var waiter in due) |
| | | 300 | | waiter.Completion.TrySetResult(run); |
| | | 301 | | return default; |
| | | 302 | | } |
| | | 303 | | |
| | | 304 | | private void MaybeCrash(string stepName, bool beforeStep, string flowId) |
| | | 305 | | { |
| | | 306 | | lock (_gate) |
| | | 307 | | { |
| | | 308 | | // A flow-scoped arm only fires on ITS run: the slot is process-wide and one-shot, so |
| | | 309 | | // an unscoped crash intended for run B could be consumed by run A (or a child flow |
| | | 310 | | // reusing the step name) reaching the step first — and B's recovery assertion then |
| | | 311 | | // passes without the recovery path ever executing. |
| | | 312 | | if (_armedCrash is not { } armed |
| | | 313 | | || armed.Step != stepName |
| | | 314 | | || armed.Before != beforeStep |
| | | 315 | | || (armed.FlowId is not null && !string.Equals(armed.FlowId, flowId, StringComparison.Ordinal))) |
| | | 316 | | { |
| | | 317 | | return; |
| | | 318 | | } |
| | | 319 | | |
| | | 320 | | _armedCrash = null; |
| | | 321 | | } |
| | | 322 | | |
| | | 323 | | throw new SimulatedCrashException(stepName, beforeStep); |
| | | 324 | | } |
| | | 325 | | |
| | | 326 | | private void Record(EventKind kind, DurableFlowStepEvent step) |
| | | 327 | | { |
| | | 328 | | var recorded = new FlowProbeEvent(kind, step); |
| | | 329 | | var list = _events.GetOrAdd(step.FlowId, static _ => []); |
| | | 330 | | Waiter[] due; |
| | | 331 | | lock (_gate) |
| | | 332 | | { |
| | | 333 | | list.Add(recorded); |
| | | 334 | | due = [.. _waiters.Where(w => w.FlowId == step.FlowId && w.Predicate(recorded))]; |
| | | 335 | | foreach (var waiter in due) |
| | | 336 | | _waiters.Remove(waiter); |
| | | 337 | | } |
| | | 338 | | |
| | | 339 | | foreach (var waiter in due) |
| | | 340 | | waiter.Completion.TrySetResult(recorded); |
| | | 341 | | } |
| | | 342 | | |
| | | 343 | | internal int CountEvents(string flowId, string stepName, EventKind kind) |
| | | 344 | | { |
| | | 345 | | if (!_events.TryGetValue(flowId, out var list)) |
| | | 346 | | return 0; |
| | | 347 | | |
| | | 348 | | lock (_gate) |
| | | 349 | | return list.Count(e => e.Kind == kind && e.Step.StepName == stepName); |
| | | 350 | | } |
| | | 351 | | |
| | | 352 | | internal IReadOnlyList<FlowProbeEvent> EventsFor(string flowId) |
| | | 353 | | { |
| | | 354 | | if (!_events.TryGetValue(flowId, out var list)) |
| | | 355 | | return []; |
| | | 356 | | |
| | | 357 | | lock (_gate) |
| | | 358 | | return [.. list]; |
| | | 359 | | } |
| | | 360 | | |
| | | 361 | | internal string? LatestAwaitedCorrelationId(string flowId, string? stepName) |
| | | 362 | | { |
| | | 363 | | lock (_gate) |
| | | 364 | | return LatestAwaitedCorrelationIdCore(flowId, stepName); |
| | | 365 | | } |
| | | 366 | | |
| | | 367 | | /// <summary>Same as <see cref="LatestAwaitedCorrelationId"/>; the caller holds <c>_gate</c>.</summary> |
| | | 368 | | private string? LatestAwaitedCorrelationIdCore(string flowId, string? stepName) |
| | | 369 | | { |
| | | 370 | | if (!_events.TryGetValue(flowId, out var list)) |
| | | 371 | | return null; |
| | | 372 | | |
| | | 373 | | // Walking backwards, remember the correlation ids of already-answered awaited steps: a |
| | | 374 | | // Waiting event whose step has a LATER Completed event is consumed — a reply published |
| | | 375 | | // to that dead correlation id is silently dropped by the channel, and the caller's |
| | | 376 | | // null-fallback (wait for the run's NEXT awaited step to park) is the correct path. |
| | | 377 | | // Progress-aware steps stay un-completed across non-terminal replies, so repeated |
| | | 378 | | // replies to the same live correlation id still resolve here. Only the NEWEST Waiting |
| | | 379 | | // of each step is considered live at all: a faulted attempt leaves no Completed event |
| | | 380 | | // behind, and returning its abandoned correlation id (an older Waiting of a step the |
| | | 381 | | // run has since restarted with a fresh id) would park the caller's reply forever. |
| | | 382 | | HashSet<string>? answered = null; |
| | | 383 | | HashSet<string>? seenWaitingSteps = null; |
| | | 384 | | for (var index = list.Count - 1; index >= 0; index--) |
| | | 385 | | { |
| | | 386 | | var candidate = list[index]; |
| | | 387 | | if (candidate.Kind == EventKind.Completed |
| | | 388 | | && candidate.Step.Kind == DurableFlowStepKind.Awaited |
| | | 389 | | && candidate.Step.CorrelationId is { } answeredCid) |
| | | 390 | | { |
| | | 391 | | (answered ??= new HashSet<string>(StringComparer.Ordinal)).Add(answeredCid); |
| | | 392 | | continue; |
| | | 393 | | } |
| | | 394 | | |
| | | 395 | | if (candidate.Kind == EventKind.Waiting |
| | | 396 | | && candidate.Step.Kind == DurableFlowStepKind.Awaited |
| | | 397 | | && candidate.Step.CorrelationId is { } correlationId) |
| | | 398 | | { |
| | | 399 | | var newestForStep = (seenWaitingSteps ??= new HashSet<string>(StringComparer.Ordinal)).Add(candidate.Ste |
| | | 400 | | if (!newestForStep || answered?.Contains(correlationId) == true) |
| | | 401 | | continue; |
| | | 402 | | |
| | | 403 | | if (stepName is null || candidate.Step.StepName == stepName) |
| | | 404 | | return correlationId; |
| | | 405 | | } |
| | | 406 | | } |
| | | 407 | | |
| | | 408 | | return null; |
| | | 409 | | } |
| | | 410 | | |
| | | 411 | | /// <summary> |
| | | 412 | | /// Returns the newest un-answered awaited-step correlation id, or waits for the run's NEXT |
| | | 413 | | /// awaited park when none is live. The liveness re-check runs under the same gate as the |
| | | 414 | | /// waiter registration, so a park recorded between the caller's |
| | | 415 | | /// <see cref="LatestAwaitedCorrelationId"/> miss and this call cannot be skipped — and, |
| | | 416 | | /// unlike a history replay, an already-answered Waiting is never returned (a reply published |
| | | 417 | | /// to a consumed correlation id is silently dropped by the channel). |
| | | 418 | | /// </summary> |
| | | 419 | | internal async Task<string> WaitForNextAwaitedAsync( |
| | | 420 | | string flowId, |
| | | 421 | | string? stepName, |
| | | 422 | | TimeSpan realTimeGuard, |
| | | 423 | | string description) |
| | | 424 | | { |
| | | 425 | | Waiter waiter; |
| | | 426 | | lock (_gate) |
| | | 427 | | { |
| | | 428 | | if (LatestAwaitedCorrelationIdCore(flowId, stepName) is { } live) |
| | | 429 | | return live; |
| | | 430 | | |
| | | 431 | | waiter = new Waiter( |
| | | 432 | | flowId, |
| | | 433 | | e => e.Kind == EventKind.Waiting |
| | | 434 | | && e.Step.Kind == DurableFlowStepKind.Awaited |
| | | 435 | | && e.Step.CorrelationId is not null |
| | | 436 | | && (stepName is null || e.Step.StepName == stepName), |
| | | 437 | | new TaskCompletionSource<FlowProbeEvent>(TaskCreationOptions.RunContinuationsAsynchronously)); |
| | | 438 | | _waiters.Add(waiter); |
| | | 439 | | } |
| | | 440 | | |
| | | 441 | | try |
| | | 442 | | { |
| | | 443 | | var stepEvent = await AwaitBoundedAsync(waiter.Completion.Task, realTimeGuard, description).ConfigureAwait(f |
| | | 444 | | return stepEvent.Step.CorrelationId!; |
| | | 445 | | } |
| | | 446 | | finally |
| | | 447 | | { |
| | | 448 | | // Same cleanup contract as WaitForAsync below: timed-out waiters must not accumulate. |
| | | 449 | | lock (_gate) |
| | | 450 | | _waiters.Remove(waiter); |
| | | 451 | | } |
| | | 452 | | } |
| | | 453 | | |
| | | 454 | | internal async Task<FlowProbeEvent> WaitForAsync( |
| | | 455 | | string flowId, |
| | | 456 | | Func<FlowProbeEvent, bool> predicate, |
| | | 457 | | TimeSpan realTimeGuard, |
| | | 458 | | string description) |
| | | 459 | | { |
| | | 460 | | Waiter waiter; |
| | | 461 | | lock (_gate) |
| | | 462 | | { |
| | | 463 | | if (_events.TryGetValue(flowId, out var list)) |
| | | 464 | | { |
| | | 465 | | // Replay NEWEST match first: a faulted attempt leaves an older Waiting event for a |
| | | 466 | | // step the run has since restarted with a fresh correlation id (nothing ever |
| | | 467 | | // Completes the abandoned one), so front-to-back replay handed back the abandoned |
| | | 468 | | // event — a reply to its correlation id was silently dropped and the live wait |
| | | 469 | | // never resolved. |
| | | 470 | | for (var index = list.Count - 1; index >= 0; index--) |
| | | 471 | | { |
| | | 472 | | if (predicate(list[index])) |
| | | 473 | | return list[index]; |
| | | 474 | | } |
| | | 475 | | } |
| | | 476 | | |
| | | 477 | | waiter = new Waiter(flowId, predicate, new TaskCompletionSource<FlowProbeEvent>(TaskCreationOptions.RunConti |
| | | 478 | | _waiters.Add(waiter); |
| | | 479 | | } |
| | | 480 | | |
| | | 481 | | try |
| | | 482 | | { |
| | | 483 | | return await AwaitBoundedAsync(waiter.Completion.Task, realTimeGuard, description).ConfigureAwait(false); |
| | | 484 | | } |
| | | 485 | | finally |
| | | 486 | | { |
| | | 487 | | // A satisfied waiter was already removed by Record; a timed-out one would otherwise |
| | | 488 | | // stay registered forever — its predicate re-evaluated on every future event and its |
| | | 489 | | // completion source retained. The gate makes the removal race-safe against a Record |
| | | 490 | | // completing it at the same moment (either way it leaves the list exactly once). |
| | | 491 | | lock (_gate) |
| | | 492 | | _waiters.Remove(waiter); |
| | | 493 | | } |
| | | 494 | | } |
| | | 495 | | |
| | | 496 | | internal async Task<DurableFlowRunEvent> WaitForRunAsync(string flowId, TimeSpan realTimeGuard, string description) |
| | | 497 | | { |
| | | 498 | | RunWaiter waiter; |
| | | 499 | | lock (_gate) |
| | | 500 | | { |
| | | 501 | | if (_finished.TryGetValue(flowId, out var finished)) |
| | | 502 | | return finished; |
| | | 503 | | |
| | | 504 | | waiter = new RunWaiter(flowId, new TaskCompletionSource<DurableFlowRunEvent>(TaskCreationOptions.RunContinua |
| | | 505 | | _runWaiters.Add(waiter); |
| | | 506 | | } |
| | | 507 | | |
| | | 508 | | try |
| | | 509 | | { |
| | | 510 | | return await AwaitBoundedAsync(waiter.Completion.Task, realTimeGuard, description).ConfigureAwait(false); |
| | | 511 | | } |
| | | 512 | | finally |
| | | 513 | | { |
| | | 514 | | // Same cleanup contract as WaitForAsync: timed-out run waiters must not accumulate. |
| | | 515 | | lock (_gate) |
| | | 516 | | _runWaiters.Remove(waiter); |
| | | 517 | | } |
| | | 518 | | } |
| | | 519 | | |
| | | 520 | | /// <summary>Registered-but-unsatisfied waiters — exposed so tests can prove timed-out waits don't leak.</summary> |
| | | 521 | | internal int PendingWaiterCount |
| | | 522 | | { |
| | | 523 | | get |
| | | 524 | | { |
| | | 525 | | lock (_gate) |
| | | 526 | | return _waiters.Count + _runWaiters.Count; |
| | | 527 | | } |
| | | 528 | | } |
| | | 529 | | |
| | | 530 | | private static async Task<T> AwaitBoundedAsync<T>(Task<T> task, TimeSpan realTimeGuard, string description) |
| | | 531 | | { |
| | | 532 | | // The guard runs on REAL time deliberately: it bounds a hung test regardless of what the |
| | | 533 | | // virtual clock is doing (which may be exactly what is broken). |
| | | 534 | | try |
| | | 535 | | { |
| | | 536 | | return await task.WaitAsync(realTimeGuard, TimeProvider.System).ConfigureAwait(false); |
| | | 537 | | } |
| | | 538 | | catch (TimeoutException) |
| | | 539 | | { |
| | | 540 | | throw new TimeoutException( |
| | | 541 | | $"Timed out after {realTimeGuard} (real time) waiting for {description}. " + |
| | | 542 | | "If the flow is sleeping on a timer or a retry backoff, advance the virtual clock first. " + |
| | | 543 | | "A run whose last execution failed and exhausted the transport's retry budget also never " + |
| | | 544 | | "finishes — it stays Running with its wake-up dropped; check the logs for 'dropping it'."); |
| | | 545 | | } |
| | | 546 | | } |
| | | 547 | | |
| | | 548 | | private sealed record Waiter(string FlowId, Func<FlowProbeEvent, bool> Predicate, TaskCompletionSource<FlowProbeEven |
| | | 549 | | |
| | | 550 | | private sealed record RunWaiter(string FlowId, TaskCompletionSource<DurableFlowRunEvent> Completion); |
| | | 551 | | } |