| | | 1 | | using Microsoft.Extensions.DependencyInjection; |
| | | 2 | | using Microsoft.Extensions.DependencyInjection.Extensions; |
| | | 3 | | using Microsoft.Extensions.Hosting; |
| | | 4 | | using Microsoft.Extensions.Logging; |
| | | 5 | | using Microsoft.Extensions.Logging.Abstractions; |
| | | 6 | | |
| | | 7 | | namespace AsyncResponse.Testing; |
| | | 8 | | |
| | | 9 | | /// <summary>Options for <see cref="AsyncResponseTestHarness.StartAsync"/>.</summary> |
| | | 10 | | public sealed class AsyncResponseTestHarnessOptions |
| | | 11 | | { |
| | | 12 | | /// <summary>The virtual clock's start instant. Default: <see cref="VirtualTimeProvider.DefaultStartTime"/>.</summar |
| | 188 | 13 | | public DateTimeOffset? StartTime { get; set; } |
| | | 14 | | |
| | | 15 | | /// <summary>Extra service registrations (fakes for the services your flows and triggers inject).</summary> |
| | 334 | 16 | | public Action<IServiceCollection>? ConfigureServices { get; set; } |
| | | 17 | | |
| | | 18 | | /// <summary> |
| | | 19 | | /// Continues the <c>AddAsyncResponse()</c> fluent chain: register flows |
| | | 20 | | /// (<c>WithDurableFlow</c>), schedules (<c>WithScheduledFlow</c>), context propagators, … |
| | | 21 | | /// The in-memory channel, transport, and flow store are already registered. |
| | | 22 | | /// </summary> |
| | 338 | 23 | | public Action<AsyncResponseRegistrationBuilder>? ConfigureAsyncResponse { get; set; } |
| | | 24 | | |
| | | 25 | | /// <summary>In-memory channel options (timeouts, recovery expiry).</summary> |
| | 200 | 26 | | public Action<InMemoryAsyncResponseOptions>? Channel { get; set; } |
| | | 27 | | |
| | | 28 | | /// <summary>In-memory worker transport options (concurrency, retry budget).</summary> |
| | 202 | 29 | | public Action<InMemoryWorkerTransportOptions>? Transport { get; set; } |
| | | 30 | | |
| | | 31 | | /// <summary>Durable-flow engine options (lease durations, step timeout, timer threshold).</summary> |
| | 218 | 32 | | public Action<DurableFlowOptions>? DurableFlows { get; set; } |
| | | 33 | | |
| | | 34 | | /// <summary> |
| | | 35 | | /// Bound on how long harness idle-waits spin in <em>real</em> time before failing the test — |
| | | 36 | | /// the safety net that turns a hang into a diagnosable failure. Default: 10 seconds. |
| | | 37 | | /// </summary> |
| | 638 | 38 | | public TimeSpan RealTimeGuard { get; set; } = TimeSpan.FromSeconds(10); |
| | | 39 | | |
| | | 40 | | /// <summary> |
| | | 41 | | /// What <see cref="AsyncResponseTestHarness.SimulateRestartAsync"/> does when user code is |
| | | 42 | | /// still executing after the old incarnation's graceful stop lapsed |
| | | 43 | | /// (<see cref="RealTimeGuard"/>): a step body that ignored its cancellation and is blocked on |
| | | 44 | | /// something the test controls, for example. A simulated restart is <b>cooperative</b> — it |
| | | 45 | | /// discards the process-bound state a crash would lose, but it cannot terminate a running |
| | | 46 | | /// delegate the way a process kill does — so such an execution would keep running beside the |
| | | 47 | | /// new incarnation and perform its side effects after the "restart" returned, proving less |
| | | 48 | | /// than the test claims. Default (<c>false</c>): the restart fails with |
| | | 49 | | /// <see cref="InvalidOperationException"/> naming the count. <c>true</c>: the executions are |
| | | 50 | | /// abandoned (their leases broken, their provider disposed) and the restart proceeds; the |
| | | 51 | | /// test then owns the overlap. Engine-owned parks — an awaited step or an in-process timer |
| | | 52 | | /// holding its worker slot on the virtual clock — are not user code and never trip this. |
| | | 53 | | /// </summary> |
| | 6 | 54 | | public bool AbandonLingeringExecutionsOnRestart { get; set; } |
| | | 55 | | |
| | | 56 | | /// <summary> |
| | | 57 | | /// Flow-execution observers installed into every incarnation (the current one and each |
| | | 58 | | /// simulated restart). <see cref="FlowTestHarness"/> installs its probe here. |
| | | 59 | | /// </summary> |
| | 486 | 60 | | public IList<IDurableFlowExecutionObserver> FlowObservers { get; } = []; |
| | | 61 | | } |
| | | 62 | | |
| | | 63 | | /// <summary> |
| | | 64 | | /// Hosts the complete AsyncResponse engine in process for tests: the in-memory channel (with full |
| | | 65 | | /// lost-subscriber recovery), the in-memory worker transport (with native delayed delivery), the |
| | | 66 | | /// in-memory durable-flow store, and every background service — all running on a |
| | | 67 | | /// <see cref="VirtualTimeProvider"/>. Production-sized timeouts, leases, timers, and cron |
| | | 68 | | /// schedules elapse only when the test calls <see cref="AdvanceAsync"/>, so nothing in a test ever |
| | | 69 | | /// sleeps for real. |
| | | 70 | | /// <para> |
| | | 71 | | /// <see cref="SimulateRestartAsync"/> models a redeploy: the service provider (and with it every |
| | | 72 | | /// live waiter and in-flight execution context) is discarded and rebuilt, while the recovery |
| | | 73 | | /// store, the flow ledgers, and scheduled (delayed) worker jobs survive — the durable state a |
| | | 74 | | /// broker- and store-backed deployment would retain. Responses published after the restart route |
| | | 75 | | /// through the real lost-subscriber recovery machinery. Waiter tasks obtained before the restart |
| | | 76 | | /// never complete (their process is gone — do not await them across a restart). |
| | | 77 | | /// </para> |
| | | 78 | | /// <para>For flow-focused tests, <see cref="FlowTestHarness"/> wraps this with step-level tooling.</para> |
| | | 79 | | /// </summary> |
| | | 80 | | public sealed class AsyncResponseTestHarness : IAsyncDisposable |
| | | 81 | | { |
| | | 82 | | private readonly AsyncResponseTestHarnessOptions _options; |
| | | 83 | | private readonly InMemoryRecoveryStateStore _recoveryStore; |
| | | 84 | | private readonly InMemoryFlowStateStore _flowStore; |
| | | 85 | | private readonly IDurableFlowExecutionObserver[] _observers; |
| | | 86 | | private readonly QuiesceProbe _quiesce = new(); |
| | | 87 | | private ServiceProvider _provider = null!; |
| | | 88 | | private IHostedService[] _started = []; |
| | | 89 | | private bool _disposed; |
| | | 90 | | |
| | | 91 | | private AsyncResponseTestHarness(AsyncResponseTestHarnessOptions options) |
| | | 92 | | { |
| | | 93 | | _options = options; |
| | | 94 | | _observers = [.. options.FlowObservers]; |
| | | 95 | | Clock = new VirtualTimeProvider(options.StartTime ?? VirtualTimeProvider.DefaultStartTime); |
| | | 96 | | _recoveryStore = new InMemoryRecoveryStateStore(Clock); |
| | | 97 | | _flowStore = new InMemoryFlowStateStore(Clock); |
| | | 98 | | } |
| | | 99 | | |
| | | 100 | | /// <summary>Builds the engine and starts its background services.</summary> |
| | | 101 | | public static async Task<AsyncResponseTestHarness> StartAsync(Action<AsyncResponseTestHarnessOptions>? configure = n |
| | | 102 | | { |
| | | 103 | | var options = new AsyncResponseTestHarnessOptions(); |
| | | 104 | | configure?.Invoke(options); |
| | | 105 | | |
| | | 106 | | var harness = new AsyncResponseTestHarness(options); |
| | | 107 | | harness.BuildProvider(); |
| | | 108 | | try |
| | | 109 | | { |
| | | 110 | | await harness.StartHostedServicesAsync().ConfigureAwait(false); |
| | | 111 | | } |
| | | 112 | | catch |
| | | 113 | | { |
| | | 114 | | await harness.DisposeAsync().ConfigureAwait(false); |
| | | 115 | | throw; |
| | | 116 | | } |
| | | 117 | | |
| | | 118 | | return harness; |
| | | 119 | | } |
| | | 120 | | |
| | | 121 | | /// <summary>The virtual clock every engine component runs on.</summary> |
| | | 122 | | public VirtualTimeProvider Clock { get; } |
| | | 123 | | |
| | | 124 | | /// <summary>The current incarnation's service provider (rebuilt by <see cref="SimulateRestartAsync"/>).</summary> |
| | | 125 | | public IServiceProvider Services => _provider; |
| | | 126 | | |
| | | 127 | | /// <summary>The fluent waiter builder (recoverable: the in-memory channel supports lost-subscriber callbacks).</sum |
| | | 128 | | public IRecoverableAsyncResponseBuilder Builder => _provider.GetRequiredService<IRecoverableAsyncResponseBuilder>(); |
| | | 129 | | |
| | | 130 | | /// <summary>The response publisher — the test's stand-in for the remote systems that answer requests.</summary> |
| | | 131 | | public IAsyncResponsePublisher Publisher => _provider.GetRequiredService<IAsyncResponsePublisher>(); |
| | | 132 | | |
| | | 133 | | /// <summary>Durable-flow starter/operations surface.</summary> |
| | | 134 | | public IDurableFlows Flows => _provider.GetRequiredService<IDurableFlows>(); |
| | | 135 | | |
| | | 136 | | /// <summary>The flow executor, for driving runs directly instead of through the worker queue.</summary> |
| | | 137 | | public IDurableFlowExecutor FlowExecutor => _provider.GetRequiredService<IDurableFlowExecutor>(); |
| | | 138 | | |
| | | 139 | | private InMemoryWorkerTransport Transport => _provider.GetRequiredService<InMemoryWorkerTransport>(); |
| | | 140 | | |
| | | 141 | | /// <summary>Publishes a response payload to a correlation id (what the remote system would do).</summary> |
| | | 142 | | public Task PublishAsync<T>(T response, string correlationId) where T : IAsyncResponsePayload |
| | | 143 | | => Publisher.SetResponse(response, correlationId); |
| | | 144 | | |
| | | 145 | | /// <summary>Publishes an exception to a correlation id (a remote failure).</summary> |
| | | 146 | | public Task PublishExceptionAsync(Exception exception, string correlationId) |
| | | 147 | | => Publisher.SetException(exception, correlationId); |
| | | 148 | | |
| | | 149 | | /// <summary> |
| | | 150 | | /// Advances virtual time by <paramref name="delta"/>, stepping timer-by-timer and letting the |
| | | 151 | | /// worker pipeline settle between steps, so work a fired timer enqueues (a durable-timer |
| | | 152 | | /// wake-up that re-arms the next chunk, a retry backoff that re-executes) is honored within |
| | | 153 | | /// this same advance. |
| | | 154 | | /// </summary> |
| | | 155 | | public async Task AdvanceAsync(TimeSpan delta) |
| | | 156 | | { |
| | | 157 | | ArgumentOutOfRangeException.ThrowIfLessThan(delta, TimeSpan.Zero); |
| | | 158 | | var target = Clock.GetUtcNow() + delta; |
| | | 159 | | |
| | | 160 | | while (true) |
| | | 161 | | { |
| | | 162 | | await SettleAsync().ConfigureAwait(false); |
| | | 163 | | |
| | | 164 | | var next = Clock.NextTimerDueAt; |
| | | 165 | | if (next is null || next > target) |
| | | 166 | | { |
| | | 167 | | Clock.AdvanceTo(target); |
| | | 168 | | break; |
| | | 169 | | } |
| | | 170 | | |
| | | 171 | | Clock.AdvanceTo(next.Value); |
| | | 172 | | } |
| | | 173 | | |
| | | 174 | | await SettleAsync().ConfigureAwait(false); |
| | | 175 | | } |
| | | 176 | | |
| | | 177 | | /// <summary> |
| | | 178 | | /// Waits (bounded by <see cref="AsyncResponseTestHarnessOptions.RealTimeGuard"/> of real time) |
| | | 179 | | /// until the in-memory worker transport has no queued or executing jobs. Do not call while a |
| | | 180 | | /// flow is parked on an in-process virtual-time wait — that job stays outstanding until the |
| | | 181 | | /// clock advances; await the flow's outcome instead. |
| | | 182 | | /// </summary> |
| | | 183 | | public async Task WaitForWorkerIdleAsync() |
| | | 184 | | { |
| | | 185 | | var guard = TimeProvider.System.GetUtcNow() + _options.RealTimeGuard; |
| | | 186 | | while (Transport.OutstandingJobs != 0) |
| | | 187 | | { |
| | | 188 | | if (TimeProvider.System.GetUtcNow() > guard) |
| | | 189 | | { |
| | | 190 | | throw new TimeoutException( |
| | | 191 | | $"The in-memory worker transport still has {Transport.OutstandingJobs} outstanding job(s) after {_op |
| | | 192 | | "A job may be parked on virtual time (advance the clock) or genuinely stuck."); |
| | | 193 | | } |
| | | 194 | | |
| | | 195 | | await Task.Delay(TimeSpan.FromMilliseconds(1)).ConfigureAwait(false); |
| | | 196 | | } |
| | | 197 | | } |
| | | 198 | | |
| | | 199 | | /// <summary> |
| | | 200 | | /// Simulates a redeploy/restart. Durable state survives: recovery registrations, flow ledgers, |
| | | 201 | | /// and scheduled (delayed) worker jobs — re-published into the new incarnation with their |
| | | 202 | | /// remaining virtual delay, as a broker would retain them. Everything process-bound dies: live |
| | | 203 | | /// waiters, subscriptions, in-flight executions (their execution leases are broken, as a real |
| | | 204 | | /// crash's silence would let them expire, so the new incarnation takes their flows over |
| | | 205 | | /// immediately). Queued immediate jobs are drained gracefully before the old incarnation |
| | | 206 | | /// stops, bounded by the real-time guard. |
| | | 207 | | /// </summary> |
| | | 208 | | /// <param name="whileDown"> |
| | | 209 | | /// Runs between the old incarnation stopping and the new one starting — with no engine |
| | | 210 | | /// running. Advance the clock here to simulate an outage: <c>Clock.Advance(TimeSpan.FromHours(4))</c> |
| | | 211 | | /// makes cron schedules skip the occurrences that fell into the downtime, exactly as a real |
| | | 212 | | /// outage would. |
| | | 213 | | /// </param> |
| | | 214 | | /// <exception cref="InvalidOperationException"> |
| | | 215 | | /// User code of the old incarnation was still executing after the graceful stop lapsed and |
| | | 216 | | /// <see cref="AsyncResponseTestHarnessOptions.AbandonLingeringExecutionsOnRestart"/> is off: |
| | | 217 | | /// the restart is cooperative and cannot kill that code, so it refuses to report a restart |
| | | 218 | | /// the surviving execution would contradict. |
| | | 219 | | /// </exception> |
| | | 220 | | public async Task SimulateRestartAsync(Action? whileDown = null) |
| | | 221 | | { |
| | | 222 | | ObjectDisposedException.ThrowIf(_disposed, this); |
| | | 223 | | |
| | | 224 | | // Retention, not a snapshot: the drain moves every pending delayed job into this list and |
| | | 225 | | // also captures delayed publishes made BY draining jobs (a flow suspending mid-drain), so |
| | | 226 | | // nothing falls between a pre-stop snapshot and the drain — a broker would keep all of it. |
| | | 227 | | var pendingDelayed = Transport.BeginRetainingDelayedJobs(); |
| | | 228 | | // Resolved BEFORE the provider goes away; abandoned after, once nothing can add to it. |
| | | 229 | | var dyingChannel = _provider.GetService<InMemoryAsyncResponseChannel>(); |
| | | 230 | | await StopHostedServicesAsync().ConfigureAwait(false); |
| | | 231 | | |
| | | 232 | | // Quiescence check BEFORE the provider is discarded. Jobs still outstanding after the |
| | | 233 | | // bounded stop are executions the stop could not end. Engine-owned parks (an awaited step |
| | | 234 | | // or an in-process timer holding its worker slot on the virtual clock) are expected — |
| | | 235 | | // their leases are broken below and the new incarnation takes them over, as after a real |
| | | 236 | | // crash. Anything beyond them is USER code still running: this restart cannot terminate |
| | | 237 | | // it (there is no process to kill), so reporting a restart while it keeps executing — |
| | | 238 | | // and performs side effects after the restart "completed" — would prove less than the |
| | | 239 | | // test claims. Refuse unless the test opted into owning that overlap. |
| | | 240 | | var lingering = Transport.OutstandingJobs + _quiesce.DirectRunsInFlight - _quiesce.ParkedCount; |
| | | 241 | | if (lingering > 0 && !_options.AbandonLingeringExecutionsOnRestart) |
| | | 242 | | { |
| | | 243 | | throw new InvalidOperationException( |
| | | 244 | | $"{nameof(SimulateRestartAsync)} could not establish quiescence: {lingering} execution(s) of the old inc |
| | | 245 | | $"were still running user code after the graceful stop lapsed ({_options.RealTimeGuard} of real time). A |
| | | 246 | | "restart is cooperative — it cannot terminate a running delegate the way a process kill does — so that c |
| | | 247 | | "keep running beside the new incarnation and perform its side effects after the restart. Let the step ob |
| | | 248 | | "cancellation token or finish before restarting, inject a crash at the checkpoint boundary with " + |
| | | 249 | | "FlowTestHarness.CrashBeforeStep/CrashAfterStep, or set " + |
| | | 250 | | $"{nameof(AsyncResponseTestHarnessOptions)}.{nameof(AsyncResponseTestHarnessOptions.AbandonLingeringExec |
| | | 251 | | "to accept the overlap."); |
| | | 252 | | } |
| | | 253 | | |
| | | 254 | | await _provider.DisposeAsync().ConfigureAwait(false); |
| | | 255 | | |
| | | 256 | | // Hard-crash semantics for whatever survived the graceful stop: a parked execution's |
| | | 257 | | // lease-renew loop runs on the SHARED virtual clock against the SHARED flow store, so it |
| | | 258 | | // would keep the lease alive forever and the new incarnation could never take the flow |
| | | 259 | | // over — the redelivered wake-up would see "executing on another live worker", ack, and |
| | | 260 | | // the flow would never resume. Breaking the leases is what a real crash's silence does |
| | | 261 | | // (the lease expires); the zombie's next renewal fails, marks itself lost, and stops. Any |
| | | 262 | | // zombie retry after that dies against its own disposed provider before touching the |
| | | 263 | | // store. The quiesce probe's parked entries died with the incarnation too. |
| | | 264 | | _flowStore.ExpireAllLeases(); |
| | | 265 | | _quiesce.Reset(); |
| | | 266 | | |
| | | 267 | | // Same hard-crash semantics for the dead incarnation's response waiters. Their timeout |
| | | 268 | | // timers were armed on the SHARED virtual clock, so without this they stayed live past the |
| | | 269 | | // "restart": advancing time fired them, completed a pre-restart ResponseTask that a crash |
| | | 270 | | // would leave hanging forever, and ran the cleanup that DELETES the registration from the |
| | | 271 | | // SHARED recovery store — so the late response this scenario exists to test found no |
| | | 272 | | // waiter AND no registration, and was silently dropped. Abandoning leaves the registration |
| | | 273 | | // exactly as a crash does: alive until its TTL, recoverable by the new incarnation. |
| | | 274 | | if (dyingChannel is not null) |
| | | 275 | | await dyingChannel.AbandonAllAsync().ConfigureAwait(false); |
| | | 276 | | |
| | | 277 | | whileDown?.Invoke(); |
| | | 278 | | |
| | | 279 | | BuildProvider(); |
| | | 280 | | await StartHostedServicesAsync().ConfigureAwait(false); |
| | | 281 | | |
| | | 282 | | if (pendingDelayed.Count > 0) |
| | | 283 | | { |
| | | 284 | | var transport = (IDelayedWorkerTransport)Transport; |
| | | 285 | | var now = Clock.GetUtcNow().UtcDateTime; |
| | | 286 | | foreach (var job in pendingDelayed) |
| | | 287 | | { |
| | | 288 | | var remaining = job.NotBeforeUtc is { } notBefore && notBefore > now |
| | | 289 | | ? notBefore - now |
| | | 290 | | : TimeSpan.Zero; |
| | | 291 | | if (remaining > TimeSpan.Zero) |
| | | 292 | | { |
| | | 293 | | // Per-hop clamp, as every production publisher applies: NotBeforeUtc rides the |
| | | 294 | | // envelope, so the executor re-delays the remainder on delivery. Unclamped, a |
| | | 295 | | // legal 60-day sleep would throw here and silently lose the rest of the list. |
| | | 296 | | var hop = remaining <= transport.MaxPublishDelay ? remaining : transport.MaxPublishDelay; |
| | | 297 | | await transport.PublishAsync(job, hop).ConfigureAwait(false); |
| | | 298 | | } |
| | | 299 | | else |
| | | 300 | | await ((IWorkerTransport)transport).PublishAsync(job).ConfigureAwait(false); |
| | | 301 | | } |
| | | 302 | | } |
| | | 303 | | } |
| | | 304 | | |
| | | 305 | | internal TimeSpan RealTimeGuard => _options.RealTimeGuard; |
| | | 306 | | |
| | | 307 | | private void BuildProvider() |
| | | 308 | | { |
| | | 309 | | var services = new ServiceCollection(); |
| | | 310 | | |
| | | 311 | | // The engine clock, the shared durable state, and the harness observers go in FIRST so the |
| | | 312 | | // TryAdd registrations inside AddAsyncResponse()/With*() adopt them. |
| | | 313 | | services.AddSingleton<TimeProvider>(Clock); |
| | | 314 | | services.AddSingleton(_recoveryStore); |
| | | 315 | | services.AddSingleton(_flowStore); |
| | | 316 | | services.AddSingleton<IDurableFlowExecutionObserver>(_quiesce); |
| | | 317 | | foreach (var observer in _observers) |
| | | 318 | | services.AddSingleton<IDurableFlowExecutionObserver>(observer); |
| | | 319 | | |
| | | 320 | | _options.ConfigureServices?.Invoke(services); |
| | | 321 | | |
| | | 322 | | // The engine resolves the LAST TimeProvider registration, so a clock registered in |
| | | 323 | | // ConfigureServices would silently displace the virtual one: no engine timer ever arms, |
| | | 324 | | // AdvanceAsync advances a clock nothing reads, and every wait dies as an unexplained |
| | | 325 | | // RealTimeGuard timeout. Fail construction instead, naming the fix. |
| | | 326 | | var lastClock = services.LastOrDefault(d => !d.IsKeyedService && d.ServiceType == typeof(TimeProvider)); |
| | | 327 | | if (lastClock is null || !ReferenceEquals(lastClock.ImplementationInstance, Clock)) |
| | | 328 | | { |
| | | 329 | | throw new InvalidOperationException( |
| | | 330 | | $"{nameof(AsyncResponseTestHarness)} drives the whole engine on its own virtual clock; a TimeProvider " |
| | | 331 | | "registered via ConfigureServices would displace it and no timer, timeout, lease, or backoff would " + |
| | | 332 | | "ever elapse. Use harness.Clock / AdvanceAsync instead of registering your own TimeProvider."); |
| | | 333 | | } |
| | | 334 | | |
| | | 335 | | // Fallback only, and only AFTER the user's registrations: AddLogging registers ILogger<> |
| | | 336 | | // with TryAdd semantics, so a non-Try registration made before ConfigureServices would |
| | | 337 | | // silently pin NullLogger and swallow the very diagnostics the harness's failure messages |
| | | 338 | | // tell users to check. |
| | | 339 | | services.TryAddSingleton(typeof(ILogger<>), typeof(NullLogger<>)); |
| | | 340 | | |
| | | 341 | | var builder = services.AddAsyncResponse() |
| | | 342 | | .WithInMemoryChannel(channel => |
| | | 343 | | { |
| | | 344 | | _options.Channel?.Invoke(channel); |
| | | 345 | | }) |
| | | 346 | | .WithInMemoryTransport(transport => |
| | | 347 | | { |
| | | 348 | | _options.Transport?.Invoke(transport); |
| | | 349 | | }) |
| | | 350 | | .WithInMemoryDurableFlows(flows => |
| | | 351 | | { |
| | | 352 | | _options.DurableFlows?.Invoke(flows); |
| | | 353 | | }); |
| | | 354 | | |
| | | 355 | | _options.ConfigureAsyncResponse?.Invoke(builder); |
| | | 356 | | |
| | | 357 | | _provider = services.BuildServiceProvider(new ServiceProviderOptions { ValidateScopes = true }); |
| | | 358 | | |
| | | 359 | | // The probe needs the engine clock and this incarnation's timer threshold to tell an |
| | | 360 | | // in-process timer park (holds a worker slot) from a suspension (the job ends) — see |
| | | 361 | | // QuiesceProbe.OnStepWaitingAsync. |
| | | 362 | | _quiesce.Arm(Clock, _provider.GetRequiredService<DurableFlowOptions>().TimerInProcessThreshold); |
| | | 363 | | } |
| | | 364 | | |
| | | 365 | | private async Task StartHostedServicesAsync() |
| | | 366 | | { |
| | | 367 | | var hosted = _provider.GetServices<IHostedService>().ToArray(); |
| | | 368 | | foreach (var service in hosted) |
| | | 369 | | await service.StartAsync(CancellationToken.None).ConfigureAwait(false); |
| | | 370 | | _started = hosted; |
| | | 371 | | } |
| | | 372 | | |
| | | 373 | | private async Task StopHostedServicesAsync() |
| | | 374 | | { |
| | | 375 | | // Bounded by the real-time guard: a graceful stop can legitimately hang when a drained job |
| | | 376 | | // is parked on virtual time; the restart then proceeds like a hard crash and the lease |
| | | 377 | | // machinery reconciles the leftover execution. |
| | | 378 | | using var cutoff = new CancellationTokenSource(_options.RealTimeGuard); |
| | | 379 | | |
| | | 380 | | // An engine-owned park — an awaited step, an in-process timer on the virtual clock — ends |
| | | 381 | | // only when the test replies or moves the clock, and neither can happen while the test is |
| | | 382 | | // awaiting this stop. Waiting it out therefore always burned the WHOLE guard (10 s by |
| | | 383 | | // default), once per restart and again per disposal, in a kit whose point is that nothing |
| | | 384 | | // sleeps for real. The moment a park is all that is left, end the wait: that is where the |
| | | 385 | | // lapsed guard was heading anyway — the leases are broken right after and the new |
| | | 386 | | // incarnation takes the execution over, exactly as after a real crash. |
| | | 387 | | using var stopWatching = new CancellationTokenSource(); |
| | | 388 | | var watcher = AbandonOnceOnlyParkedAsync(cutoff, stopWatching.Token); |
| | | 389 | | try |
| | | 390 | | { |
| | | 391 | | foreach (var service in Enumerable.Reverse(_started)) |
| | | 392 | | { |
| | | 393 | | try |
| | | 394 | | { |
| | | 395 | | await service.StopAsync(cutoff.Token).ConfigureAwait(false); |
| | | 396 | | } |
| | | 397 | | catch (OperationCanceledException) |
| | | 398 | | { |
| | | 399 | | // Hard-crash semantics for whatever did not stop in time. |
| | | 400 | | } |
| | | 401 | | } |
| | | 402 | | } |
| | | 403 | | finally |
| | | 404 | | { |
| | | 405 | | await stopWatching.CancelAsync().ConfigureAwait(false); |
| | | 406 | | await watcher.ConfigureAwait(false); |
| | | 407 | | } |
| | | 408 | | |
| | | 409 | | _started = []; |
| | | 410 | | } |
| | | 411 | | |
| | | 412 | | /// <summary> |
| | | 413 | | /// Cancels <paramref name="cutoff"/> as soon as every execution still outstanding is an |
| | | 414 | | /// engine-owned park. Deliberately does nothing while NOTHING is parked: a stop that can still |
| | | 415 | | /// make progress is left to finish cleanly, and the real-time guard stays the backstop for |
| | | 416 | | /// user code that is genuinely stuck. |
| | | 417 | | /// </summary> |
| | | 418 | | private async Task AbandonOnceOnlyParkedAsync(CancellationTokenSource cutoff, CancellationToken stopWatching) |
| | | 419 | | { |
| | | 420 | | try |
| | | 421 | | { |
| | | 422 | | while (!stopWatching.IsCancellationRequested) |
| | | 423 | | { |
| | | 424 | | var parked = _quiesce.ParkedCount; |
| | | 425 | | if (parked > 0 && Transport.OutstandingJobs + _quiesce.DirectRunsInFlight <= parked) |
| | | 426 | | { |
| | | 427 | | await cutoff.CancelAsync().ConfigureAwait(false); |
| | | 428 | | return; |
| | | 429 | | } |
| | | 430 | | |
| | | 431 | | // The SYSTEM clock: the harness's virtual one is not moving while a test awaits |
| | | 432 | | // this stop, which is the whole reason the park cannot end by itself. |
| | | 433 | | await Task.Delay(ParkedStopPollInterval, TimeProvider.System, stopWatching).ConfigureAwait(false); |
| | | 434 | | } |
| | | 435 | | } |
| | | 436 | | catch (OperationCanceledException) |
| | | 437 | | { |
| | | 438 | | // The stop finished on its own. |
| | | 439 | | } |
| | | 440 | | } |
| | | 441 | | |
| | | 442 | | /// <summary>How often the stop checks whether a park is all that is left.</summary> |
| | | 443 | | private static readonly TimeSpan ParkedStopPollInterval = TimeSpan.FromMilliseconds(5); |
| | | 444 | | |
| | | 445 | | /// <inheritdoc/> |
| | | 446 | | public async ValueTask DisposeAsync() |
| | | 447 | | { |
| | | 448 | | if (_disposed) |
| | | 449 | | return; |
| | | 450 | | |
| | | 451 | | _disposed = true; |
| | | 452 | | await StopHostedServicesAsync().ConfigureAwait(false); |
| | | 453 | | await _provider.DisposeAsync().ConfigureAwait(false); |
| | | 454 | | } |
| | | 455 | | |
| | | 456 | | /// <summary> |
| | | 457 | | /// Give worker jobs that are still RUNNING code a real chance to reach their next stable |
| | | 458 | | /// state before the clock moves. Advancing while a job is mid-code would let a later |
| | | 459 | | /// <c>DelayAsync</c> or deadline be computed from the already-advanced clock and park beyond |
| | | 460 | | /// the advance target — the load-dependent flake this settle exists to prevent. Three |
| | | 461 | | /// signals, cheapest first: |
| | | 462 | | /// <list type="bullet"> |
| | | 463 | | /// <item>Every outstanding job is parked on an event-visible engine wait (an awaited reply, |
| | | 464 | | /// an in-process flow timer) — the quiesce probe's count covers the outstanding count, and |
| | | 465 | | /// settling costs nothing.</item> |
| | | 466 | | /// <item>A virtual timer was armed while settling — the busy job just began a virtual-time |
| | | 467 | | /// wait (a retry backoff, a lease-acquisition poll, a timer chunk), so advancing the clock is |
| | | 468 | | /// exactly how it progresses.</item> |
| | | 469 | | /// <item>A bounded real-time grace for what cannot be attributed (a job blocked on a virtual |
| | | 470 | | /// timer it armed before this settle began looks identical to one stuck in user code). The |
| | | 471 | | /// budget is generous relative to scheduling noise; a fake that sleeps on the SYSTEM clock |
| | | 472 | | /// longer than this is the documented harness anti-pattern (use the injected TimeProvider).</item> |
| | | 473 | | /// </list> |
| | | 474 | | /// </summary> |
| | | 475 | | /// <summary>Reports an inline executor attempt starting (see QuiesceProbe.DirectRunsInFlight).</summary> |
| | | 476 | | internal void OnDirectRunStarted() => _quiesce.OnDirectRunStarted(); |
| | | 477 | | |
| | | 478 | | /// <summary>Reports an inline executor attempt finished.</summary> |
| | | 479 | | internal void OnDirectRunFinished() => _quiesce.OnDirectRunFinished(); |
| | | 480 | | |
| | | 481 | | private async Task SettleAsync() |
| | | 482 | | { |
| | | 483 | | // Let just-released continuations (a fired timer's write, worker dispatch) reach the |
| | | 484 | | // transport counters before reading them. |
| | | 485 | | for (var round = 0; round < 3; round++) |
| | | 486 | | { |
| | | 487 | | await Task.Yield(); |
| | | 488 | | await Task.Delay(TimeSpan.FromMilliseconds(1)).ConfigureAwait(false); |
| | | 489 | | } |
| | | 490 | | |
| | | 491 | | var budget = TimeProvider.System.GetUtcNow() + TimeSpan.FromMilliseconds(500); |
| | | 492 | | while (Transport.OutstandingJobs + _quiesce.DirectRunsInFlight > _quiesce.ParkedCount |
| | | 493 | | && TimeProvider.System.GetUtcNow() < budget) |
| | | 494 | | { |
| | | 495 | | var next = Clock.NextTimerDueAt; |
| | | 496 | | await Task.Delay(TimeSpan.FromMilliseconds(1)).ConfigureAwait(false); |
| | | 497 | | if (next != Clock.NextTimerDueAt) |
| | | 498 | | return; // A virtual wait just began — the advance loop re-evaluates immediately. |
| | | 499 | | } |
| | | 500 | | } |
| | | 501 | | |
| | | 502 | | /// <summary> |
| | | 503 | | /// Tracks flow executions parked on an engine-owned wait: between a step's Waiting event and |
| | | 504 | | /// its Completed event the execution holds a worker slot but is idle by design, waiting on |
| | | 505 | | /// virtual time or a test-published reply. Only waits that actually park in process count — |
| | | 506 | | /// child-flow steps always suspend (the job ends), and a timer whose remainder exceeds the |
| | | 507 | | /// in-process threshold suspends too on this delayed-capable transport, so its Waiting is not |
| | | 508 | | /// counted (a stale entry would outlive the worker slot until the wake-up replay and mask a |
| | | 509 | | /// genuinely busy job in SettleAsync's guard). Cleared on restart: the old incarnation's |
| | | 510 | | /// parked executions die with it. |
| | | 511 | | /// </summary> |
| | | 512 | | private sealed class QuiesceProbe : IDurableFlowExecutionObserver |
| | | 513 | | { |
| | | 514 | | private readonly HashSet<(string FlowId, string Step)> _parked = []; |
| | | 515 | | private TimeProvider _clock = TimeProvider.System; |
| | | 516 | | private TimeSpan _timerInProcessThreshold; |
| | | 517 | | private int _directRunsInFlight; |
| | | 518 | | |
| | | 519 | | public int ParkedCount |
| | | 520 | | { |
| | | 521 | | get { lock (_parked) return _parked.Count; } |
| | | 522 | | } |
| | | 523 | | |
| | | 524 | | /// <summary> |
| | | 525 | | /// Executor attempts running inline (<c>FlowRunHandle.ExecuteDirectAsync</c>), which hold |
| | | 526 | | /// no worker slot: SettleAsync counts them beside <c>Transport.OutstandingJobs</c>, since |
| | | 527 | | /// a step they park would otherwise make ParkedCount exceed the outstanding jobs and let |
| | | 528 | | /// the clock advance under a direct run still executing user code. |
| | | 529 | | /// </summary> |
| | | 530 | | public int DirectRunsInFlight => Volatile.Read(ref _directRunsInFlight); |
| | | 531 | | |
| | | 532 | | public void OnDirectRunStarted() => Interlocked.Increment(ref _directRunsInFlight); |
| | | 533 | | |
| | | 534 | | public void OnDirectRunFinished() => Interlocked.Decrement(ref _directRunsInFlight); |
| | | 535 | | |
| | | 536 | | /// <summary>Binds the engine clock and timer threshold of the current incarnation.</summary> |
| | | 537 | | public void Arm(TimeProvider clock, TimeSpan timerInProcessThreshold) |
| | | 538 | | { |
| | | 539 | | _clock = clock; |
| | | 540 | | _timerInProcessThreshold = timerInProcessThreshold; |
| | | 541 | | } |
| | | 542 | | |
| | | 543 | | public void Reset() |
| | | 544 | | { |
| | | 545 | | lock (_parked) _parked.Clear(); |
| | | 546 | | Volatile.Write(ref _directRunsInFlight, 0); |
| | | 547 | | } |
| | | 548 | | |
| | | 549 | | public ValueTask OnStepWaitingAsync(DurableFlowStepEvent step) |
| | | 550 | | { |
| | | 551 | | // Awaited steps always park in process holding their worker slot. Timer steps only |
| | | 552 | | // park when the remainder is at or below the in-process threshold — a longer wait |
| | | 553 | | // suspends (the engine mirrors this decision in DelayCoreAsync against the same |
| | | 554 | | // clock), ending the worker job this entry would otherwise be offsetting. |
| | | 555 | | var parksInProcess = step.Kind switch |
| | | 556 | | { |
| | | 557 | | DurableFlowStepKind.Awaited => true, |
| | | 558 | | DurableFlowStepKind.Timer => step.WakeAtUtc is { } wakeAtUtc |
| | | 559 | | && wakeAtUtc - _clock.GetUtcNow().UtcDateTime <= _timerInProcessThreshold, |
| | | 560 | | _ => false |
| | | 561 | | }; |
| | | 562 | | |
| | | 563 | | if (parksInProcess) |
| | | 564 | | lock (_parked) _parked.Add((step.FlowId, step.StepName)); |
| | | 565 | | return default; |
| | | 566 | | } |
| | | 567 | | |
| | | 568 | | public ValueTask OnStepCompletedAsync(DurableFlowStepEvent step) |
| | | 569 | | { |
| | | 570 | | lock (_parked) _parked.Remove((step.FlowId, step.StepName)); |
| | | 571 | | return default; |
| | | 572 | | } |
| | | 573 | | |
| | | 574 | | public ValueTask OnRunFinishedAsync(DurableFlowRunEvent run) |
| | | 575 | | { |
| | | 576 | | lock (_parked) _parked.RemoveWhere(entry => string.Equals(entry.FlowId, run.FlowId, StringComparison.Ordinal |
| | | 577 | | return default; |
| | | 578 | | } |
| | | 579 | | |
| | | 580 | | public ValueTask OnRunAttemptFailedAsync(DurableFlowRunEvent run) |
| | | 581 | | { |
| | | 582 | | // The failed attempt released its worker slot with its waits unresolved (a waiter |
| | | 583 | | // timeout, a published failure the flow does not treat as terminal); the redelivery |
| | | 584 | | // re-parks whatever still applies. Left in place, the stale entry offset a genuinely |
| | | 585 | | // busy job in SettleAsync's guard for the rest of the incarnation. |
| | | 586 | | lock (_parked) _parked.RemoveWhere(entry => string.Equals(entry.FlowId, run.FlowId, StringComparison.Ordinal |
| | | 587 | | return default; |
| | | 588 | | } |
| | | 589 | | } |
| | | 590 | | } |