| | | 1 | | using Microsoft.Extensions.DependencyInjection; |
| | | 2 | | using Microsoft.Extensions.Logging; |
| | | 3 | | using System.Diagnostics.CodeAnalysis; |
| | | 4 | | |
| | | 5 | | namespace AsyncResponse; |
| | | 6 | | |
| | | 7 | | /// <summary> |
| | | 8 | | /// Executes durable flow runs. Its methods are the durable targets behind every flow: worker jobs |
| | | 9 | | /// carry <see cref="ExecuteAsync"/>, and awaited steps register <see cref="RecoverAsync"/> / |
| | | 10 | | /// <see cref="FailAsync"/> as their lost-subscriber callbacks — invoked by whichever process |
| | | 11 | | /// receives a late response, possibly a different deployment. |
| | | 12 | | /// <para> |
| | | 13 | | /// <b>Naming contract:</b> like all recovery callbacks, these targets are persisted as |
| | | 14 | | /// interface/method name strings and live in stores for up to the configured expiry. The |
| | | 15 | | /// interface and method names must stay stable across deployments. |
| | | 16 | | /// </para> |
| | | 17 | | /// </summary> |
| | | 18 | | public interface IDurableFlowExecutor |
| | | 19 | | { |
| | | 20 | | /// <summary> |
| | | 21 | | /// Runs the flow body for <paramref name="flowId"/> from the top: completed steps skip via |
| | | 22 | | /// their checkpoints, the in-flight awaited step re-attaches. No-op for terminal runs. |
| | | 23 | | /// </summary> |
| | | 24 | | Task ExecuteAsync(string flowId); |
| | | 25 | | |
| | | 26 | | /// <summary> |
| | | 27 | | /// Lost-subscriber resume target: re-enqueues <see cref="ExecuteAsync"/> on the worker |
| | | 28 | | /// transport (never runs the flow inline on a publisher's dispatch path). |
| | | 29 | | /// </summary> |
| | | 30 | | Task ResumeAsync(string flowId); |
| | | 31 | | |
| | | 32 | | /// <summary> |
| | | 33 | | /// Lost-subscriber success target: checkpoints the terminal payload into the matching pending |
| | | 34 | | /// step before re-enqueueing execution, so recovery does not wait for a consumed correlation id. |
| | | 35 | | /// </summary> |
| | | 36 | | Task RecoverAsync(string flowId, object payload, string correlationId); |
| | | 37 | | |
| | | 38 | | /// <summary>Lost-subscriber failure target: marks the run terminally <see cref="FlowRunStatus.Failed"/>.</summary> |
| | | 39 | | Task FailAsync(string flowId, Exception exception); |
| | | 40 | | } |
| | | 41 | | |
| | | 42 | | /// <inheritdoc cref="IDurableFlowExecutor" /> |
| | | 43 | | internal sealed class DurableFlowExecutor : IDurableFlowExecutor |
| | | 44 | | { |
| | | 45 | | private readonly IServiceScopeFactory _scopeFactory; |
| | | 46 | | private readonly IAsyncResponseBuilder _builder; |
| | | 47 | | private readonly IAsyncResponseSubscriber _subscriber; |
| | | 48 | | private readonly IRecoverableAsyncResponseSubscriber? _recoverableSubscriber; |
| | | 49 | | private readonly AsyncResponseContextPropagation _propagation; |
| | | 50 | | private readonly DurableFlowOptions _options; |
| | | 51 | | private readonly ILogger<DurableFlowExecutor> _logger; |
| | | 52 | | private readonly Dictionary<string, DurableFlowRegistration> _registrations; |
| | | 53 | | private readonly CancellationToken _hostStopping; |
| | | 54 | | |
| | | 55 | | /// <summary>Creates the flow executor.</summary> |
| | 3 | 56 | | public DurableFlowExecutor( |
| | 3 | 57 | | IServiceScopeFactory scopeFactory, |
| | 3 | 58 | | IAsyncResponseBuilder builder, |
| | 3 | 59 | | IAsyncResponseSubscriber subscriber, |
| | 3 | 60 | | IRecoverableAsyncResponseSubscriber? recoverableSubscriber, |
| | 3 | 61 | | AsyncResponseContextPropagation propagation, |
| | 3 | 62 | | DurableFlowOptions options, |
| | 3 | 63 | | ILogger<DurableFlowExecutor> logger, |
| | 3 | 64 | | IEnumerable<DurableFlowRegistration>? registrations = null, |
| | 3 | 65 | | Microsoft.Extensions.Hosting.IHostApplicationLifetime? hostLifetime = null) |
| | | 66 | | { |
| | 3 | 67 | | _scopeFactory = scopeFactory; |
| | 3 | 68 | | _builder = builder; |
| | 3 | 69 | | _subscriber = subscriber; |
| | 3 | 70 | | _recoverableSubscriber = recoverableSubscriber; |
| | 3 | 71 | | _propagation = propagation; |
| | 3 | 72 | | _options = options; |
| | 3 | 73 | | FlowStateConcurrency.ValidateOptions(_options); |
| | 3 | 74 | | _logger = logger; |
| | 3 | 75 | | _hostStopping = hostLifetime?.ApplicationStopping ?? CancellationToken.None; |
| | 3 | 76 | | _registrations = new Dictionary<string, DurableFlowRegistration>(StringComparer.Ordinal); |
| | 3 | 77 | | foreach (var registration in registrations ?? []) |
| | | 78 | | { |
| | | 79 | | // Last registration wins, matching DI's usual override semantics. |
| | 1 | 80 | | _registrations[registration.FlowTypeFullName] = registration; |
| | | 81 | | } |
| | 3 | 82 | | } |
| | | 83 | | |
| | | 84 | | /// <inheritdoc /> |
| | | 85 | | public async Task ExecuteAsync(string flowId) |
| | | 86 | | { |
| | 3 | 87 | | ArgumentException.ThrowIfNullOrWhiteSpace(flowId); |
| | | 88 | | |
| | 3 | 89 | | await using var scope = _scopeFactory.CreateAsyncScope(); |
| | 3 | 90 | | var store = scope.ServiceProvider.GetRequiredService<IFlowStateStore>(); |
| | | 91 | | |
| | 3 | 92 | | await using var lease = await AcquireExecutionLeaseWithRetryAsync(store, flowId).ConfigureAwait(false); |
| | 3 | 93 | | if (lease is null) |
| | | 94 | | return; |
| | | 95 | | |
| | 3 | 96 | | var state = await store.LoadAsync(flowId).ConfigureAwait(false); |
| | 3 | 97 | | if (state is null) |
| | | 98 | | { |
| | 2 | 99 | | _logger.LogWarning("Durable flow {FlowId} has no state (unknown, expired, or unreadable); nothing to execute |
| | 2 | 100 | | return; |
| | | 101 | | } |
| | | 102 | | |
| | 3 | 103 | | if (state.Status != FlowRunStatus.Running) |
| | | 104 | | { |
| | 3 | 105 | | _logger.LogDebug("Durable flow {FlowId} is already {Status}; skipping execution.", flowId, state.Status); |
| | 3 | 106 | | await NotifyParentAsync(state).ConfigureAwait(false); |
| | 3 | 107 | | return; |
| | | 108 | | } |
| | | 109 | | |
| | 3 | 110 | | using var activity = AsyncResponseDiagnostics.StartActivity("asyncresponse.flow.execute"); |
| | 3 | 111 | | activity?.SetTag("asyncresponse.flow_id", flowId); |
| | 3 | 112 | | activity?.SetTag("asyncresponse.flow_type", state.FlowTypeName); |
| | | 113 | | |
| | 3 | 114 | | state.Attempts++; |
| | 3 | 115 | | await lease.SaveAsync(state, _options.StateExpiry).ConfigureAwait(false); |
| | | 116 | | |
| | | 117 | | // The run may be resumed by a different deployment than the one that started it: restore |
| | | 118 | | // the ambient context captured at start before any flow code runs. |
| | 3 | 119 | | using var ambientScope = _propagation.Restore(state.Context); |
| | | 120 | | |
| | | 121 | | try |
| | | 122 | | { |
| | 3 | 123 | | var suspended = await InvokeFlowAsync(scope.ServiceProvider, store, state, lease).ConfigureAwait(false); |
| | 3 | 124 | | if (suspended) |
| | | 125 | | { |
| | | 126 | | // The context persisted the suspended state BEFORE enqueueing the child; saving here |
| | | 127 | | // could overwrite newer checkpoints written by a parent re-execution the child has |
| | | 128 | | // already triggered on another worker. |
| | 2 | 129 | | _logger.LogDebug("Durable flow {FlowId} suspended: {Message}", flowId, state.LastMessage); |
| | 2 | 130 | | return; |
| | | 131 | | } |
| | | 132 | | |
| | 3 | 133 | | state.Status = FlowRunStatus.Succeeded; |
| | 3 | 134 | | state.LastMessage = "Flow completed."; |
| | 3 | 135 | | await lease.SaveAsync(state, _options.StateExpiry).ConfigureAwait(false); |
| | | 136 | | |
| | 3 | 137 | | _logger.LogInformation("Durable flow {FlowId} completed successfully (attempt {Attempts}).", flowId, state.A |
| | 3 | 138 | | } |
| | 2 | 139 | | catch (DurableFlowSuspendedException ex) |
| | | 140 | | { |
| | | 141 | | // Same as the IsSuspended return above: the suspended state is already persisted, and a |
| | | 142 | | // save here races the child-triggered parent re-execution. |
| | 2 | 143 | | _logger.LogDebug("Durable flow {FlowId} suspended: {Message}", flowId, ex.Message); |
| | 2 | 144 | | return; |
| | | 145 | | } |
| | 3 | 146 | | catch (DurableFlowFailedException ex) |
| | | 147 | | { |
| | | 148 | | // Terminal by declaration: mark failed and swallow so the transport acks the job. |
| | 3 | 149 | | state.Status = FlowRunStatus.Failed; |
| | 3 | 150 | | state.LastMessage = ex.Message; |
| | 3 | 151 | | await lease.SaveAsync(state, _options.StateExpiry, cause: ex).ConfigureAwait(false); |
| | | 152 | | |
| | 3 | 153 | | AsyncResponseDiagnostics.SetError(activity, ex); |
| | 3 | 154 | | _logger.LogWarning(ex, "Durable flow {FlowId} failed terminally: {Message}", flowId, ex.Message); |
| | 3 | 155 | | } |
| | 2 | 156 | | catch (Exception ex) when (lease.LostToken.IsCancellationRequested) |
| | | 157 | | { |
| | 2 | 158 | | AsyncResponseDiagnostics.SetError(activity, ex); |
| | 2 | 159 | | throw; |
| | | 160 | | } |
| | 2 | 161 | | catch (Exception ex) |
| | | 162 | | { |
| | 2 | 163 | | state.LastMessage = ex.Message; |
| | 2 | 164 | | await lease.SaveAsync(state, _options.StateExpiry, cause: ex).ConfigureAwait(false); |
| | | 165 | | |
| | 2 | 166 | | AsyncResponseDiagnostics.SetError(activity, ex); |
| | | 167 | | |
| | | 168 | | // Retriable: propagate so the worker transport redelivers the run with bounded |
| | | 169 | | // attempts and dead-letters it when they are exhausted — the "run is stuck" alarm. |
| | 2 | 170 | | throw; |
| | 0 | 171 | | } |
| | | 172 | | |
| | 3 | 173 | | await NotifyParentAsync(state).ConfigureAwait(false); |
| | 3 | 174 | | } |
| | | 175 | | |
| | | 176 | | /// <summary> |
| | | 177 | | /// Acquires the execution lease for <paramref name="flowId"/>, retrying while the current |
| | | 178 | | /// holder's lease window elapses. Returns <c>null</c> when this delivery is safe to ack without |
| | | 179 | | /// executing (flow terminal/absent, or the lease is held by a demonstrably live worker). |
| | | 180 | | /// <para> |
| | | 181 | | /// A held lease alone is NOT proof this delivery is a duplicate: the holder may have died |
| | | 182 | | /// inside its unexpired lease window, and acking would drop the only wake-up the flow has — |
| | | 183 | | /// wake-ups would silently become at-most-once and the <see cref="FlowRunStatus.Running"/> run |
| | | 184 | | /// would strand. A dead holder's lease expires within <see cref="DurableFlowOptions.ExecutionLeaseDuration"/>, |
| | | 185 | | /// so polling for one full duration + renew interval guarantees this delivery either takes over |
| | | 186 | | /// (checkpoints make the re-run idempotent) or proves the holder alive. |
| | | 187 | | /// </para> |
| | | 188 | | /// </summary> |
| | | 189 | | private async Task<FlowExecutionLease?> AcquireExecutionLeaseWithRetryAsync(IFlowStateStore store, string flowId) |
| | | 190 | | { |
| | 3 | 191 | | var lease = await FlowStateConcurrency.TryAcquireExecutionLeaseAsync( |
| | 3 | 192 | | store, |
| | 3 | 193 | | flowId, |
| | 3 | 194 | | _options, |
| | 3 | 195 | | _logger).ConfigureAwait(false); |
| | 3 | 196 | | if (lease is not null) |
| | 3 | 197 | | return lease; |
| | | 198 | | |
| | | 199 | | // Poll ceiling: a lease can only stay held past its duration if the holder renewed it, so |
| | | 200 | | // one full duration + renew interval of failed acquires proves the holder is alive. The 2s |
| | | 201 | | // poll delay is capped by the renew interval so short test-sized leases still get polled. |
| | 2 | 202 | | var deadline = DateTime.UtcNow + _options.ExecutionLeaseDuration + _options.ExecutionLeaseRenewInterval; |
| | 2 | 203 | | var pollDelay = _options.ExecutionLeaseRenewInterval < TimeSpan.FromSeconds(2) |
| | 2 | 204 | | ? _options.ExecutionLeaseRenewInterval |
| | 2 | 205 | | : TimeSpan.FromSeconds(2); |
| | | 206 | | |
| | | 207 | | while (true) |
| | | 208 | | { |
| | | 209 | | // Between attempts, look at the state itself: a terminal or absent flow needs no |
| | | 210 | | // execution, and reporting it accurately beats a misleading "already executing" log. |
| | 2 | 211 | | var state = await store.LoadAsync(flowId).ConfigureAwait(false); |
| | 2 | 212 | | if (state is null) |
| | | 213 | | { |
| | 0 | 214 | | _logger.LogWarning("Durable flow {FlowId} has no state (unknown, expired, or unreadable); nothing to exe |
| | 0 | 215 | | return null; |
| | | 216 | | } |
| | | 217 | | |
| | 2 | 218 | | if (state.Status != FlowRunStatus.Running) |
| | | 219 | | { |
| | 2 | 220 | | _logger.LogDebug("Durable flow {FlowId} is already {Status}; skipping duplicate delivery.", flowId, stat |
| | 2 | 221 | | await NotifyParentAsync(state).ConfigureAwait(false); |
| | 2 | 222 | | return null; |
| | | 223 | | } |
| | | 224 | | |
| | 2 | 225 | | lease = await FlowStateConcurrency.TryAcquireExecutionLeaseAsync( |
| | 2 | 226 | | store, |
| | 2 | 227 | | flowId, |
| | 2 | 228 | | _options, |
| | 2 | 229 | | _logger).ConfigureAwait(false); |
| | 2 | 230 | | if (lease is not null) |
| | 2 | 231 | | return lease; |
| | | 232 | | |
| | 2 | 233 | | if (DateTime.UtcNow >= deadline) |
| | | 234 | | break; |
| | | 235 | | |
| | | 236 | | try |
| | | 237 | | { |
| | 2 | 238 | | await Task.Delay(pollDelay, _hostStopping).ConfigureAwait(false); |
| | 2 | 239 | | } |
| | 0 | 240 | | catch (OperationCanceledException) |
| | | 241 | | { |
| | | 242 | | // Host shutdown must not leave this delivery parked in the poll — but acking it |
| | | 243 | | // would silently drop the flow's only wake-up. Propagate as cancellation so the |
| | | 244 | | // transport treats the job as not executed and redelivers it after restart. |
| | 0 | 245 | | throw new OperationCanceledException( |
| | 0 | 246 | | $"Host is stopping; durable flow '{flowId}' wake-up is abandoned for redelivery."); |
| | | 247 | | } |
| | | 248 | | } |
| | | 249 | | |
| | | 250 | | // The lease survived a full duration + renew window, so the holder is alive and renewing. |
| | | 251 | | // Every execution is driven by a worker job that stays unacked until its handler completes, |
| | | 252 | | // so if that live holder crashes later the broker redelivers its own job — this delivery is |
| | | 253 | | // genuinely redundant and safe to ack. The retry loop above exists purely to cover |
| | | 254 | | // deliveries that arrive inside a DEAD holder's unexpired lease window, which broker |
| | | 255 | | // redelivery alone cannot cover. |
| | 2 | 256 | | _logger.LogDebug( |
| | 2 | 257 | | "Durable flow {FlowId} is executing on another live worker (lease renewed through the full wait window); ski |
| | 2 | 258 | | flowId); |
| | 2 | 259 | | return null; |
| | 3 | 260 | | } |
| | | 261 | | |
| | | 262 | | /// <inheritdoc /> |
| | | 263 | | public async Task ResumeAsync(string flowId) |
| | | 264 | | { |
| | 2 | 265 | | ArgumentException.ThrowIfNullOrWhiteSpace(flowId); |
| | | 266 | | |
| | 2 | 267 | | await using var scope = _scopeFactory.CreateAsyncScope(); |
| | 2 | 268 | | var store = scope.ServiceProvider.GetRequiredService<IFlowStateStore>(); |
| | | 269 | | |
| | 2 | 270 | | var state = await store.LoadAsync(flowId).ConfigureAwait(false); |
| | 2 | 271 | | if (state is null) |
| | | 272 | | { |
| | 2 | 273 | | _logger.LogWarning("Durable flow {FlowId} cannot resume: no state (unknown, expired, or unreadable).", flowI |
| | 2 | 274 | | return; |
| | | 275 | | } |
| | | 276 | | |
| | 2 | 277 | | if (state.Status != FlowRunStatus.Running) |
| | | 278 | | { |
| | 2 | 279 | | _logger.LogDebug("Durable flow {FlowId} is already {Status}; ignoring resume.", flowId, state.Status); |
| | 2 | 280 | | return; |
| | | 281 | | } |
| | | 282 | | |
| | 2 | 283 | | _logger.LogDebug("Durable flow {FlowId} resuming via worker transport.", flowId); |
| | 2 | 284 | | await _builder.EnqueueWorkerAsync<IDurableFlowExecutor>(executor => executor.ExecuteAsync(flowId)).ConfigureAwai |
| | 2 | 285 | | } |
| | | 286 | | |
| | | 287 | | /// <inheritdoc /> |
| | | 288 | | public async Task RecoverAsync(string flowId, object payload, string correlationId) |
| | | 289 | | { |
| | 2 | 290 | | ArgumentException.ThrowIfNullOrWhiteSpace(flowId); |
| | 2 | 291 | | ArgumentNullException.ThrowIfNull(payload); |
| | 2 | 292 | | ArgumentException.ThrowIfNullOrWhiteSpace(correlationId); |
| | | 293 | | |
| | 2 | 294 | | await using var scope = _scopeFactory.CreateAsyncScope(); |
| | 2 | 295 | | var store = scope.ServiceProvider.GetRequiredService<IFlowStateStore>(); |
| | 2 | 296 | | var checkpointed = false; |
| | 2 | 297 | | var running = false; |
| | 2 | 298 | | var lastStatus = FlowRunStatus.Running; |
| | | 299 | | |
| | 2 | 300 | | var found = await FlowStateConcurrency.MutateAsync( |
| | 2 | 301 | | store, |
| | 2 | 302 | | flowId, |
| | 2 | 303 | | _options.StateExpiry, |
| | 2 | 304 | | state => |
| | 2 | 305 | | { |
| | 2 | 306 | | checkpointed = false; |
| | 2 | 307 | | lastStatus = state.Status; |
| | 2 | 308 | | running = state.Status == FlowRunStatus.Running; |
| | 2 | 309 | | if (!running || state.Steps is null) |
| | 2 | 310 | | return false; |
| | 2 | 311 | | |
| | 2 | 312 | | var pending = state.Steps.FirstOrDefault(pair => |
| | 2 | 313 | | string.Equals(pair.Value.PendingCorrelationId, correlationId, StringComparison.Ordinal)); |
| | 2 | 314 | | if (pending.Value is null) |
| | 2 | 315 | | return false; |
| | 2 | 316 | | |
| | 2 | 317 | | pending.Value.Completed = true; |
| | 2 | 318 | | pending.Value.ResultJson = AsyncResponseJson.Serialize(payload, payload.GetType()); |
| | 2 | 319 | | pending.Value.PendingCorrelationId = null; |
| | 2 | 320 | | pending.Value.Faulted = false; |
| | 2 | 321 | | pending.Value.Message = "Terminal response recovered after subscriber loss."; |
| | 2 | 322 | | pending.Value.CompletedAtUtc = DateTime.UtcNow; |
| | 2 | 323 | | state.LastMessage = $"Step '{pending.Key}' recovered after subscriber loss."; |
| | 2 | 324 | | checkpointed = true; |
| | 2 | 325 | | return true; |
| | 2 | 326 | | }).ConfigureAwait(false); |
| | | 327 | | |
| | 2 | 328 | | if (!found) |
| | | 329 | | { |
| | 2 | 330 | | _logger.LogWarning("Durable flow {FlowId} cannot recover response {CorrelationId}: no state found.", flowId, |
| | 2 | 331 | | return; |
| | | 332 | | } |
| | | 333 | | |
| | 2 | 334 | | if (!checkpointed) |
| | | 335 | | { |
| | 2 | 336 | | if (!running) |
| | | 337 | | { |
| | 2 | 338 | | _logger.LogDebug("Durable flow {FlowId} is {Status}; ignoring recovered correlationId {CorrelationId}.", |
| | 2 | 339 | | return; |
| | | 340 | | } |
| | | 341 | | |
| | | 342 | | // Still Running with no matching pending step: a previous delivery may have crashed |
| | | 343 | | // between checkpointing this response and enqueueing the run, making this redelivery |
| | | 344 | | // the only remaining wake-up. Re-enqueue instead of dropping — it is idempotent, and |
| | | 345 | | // worst case the job finds a live holder's lease and acks as a duplicate. |
| | 2 | 346 | | _logger.LogDebug("Durable flow {FlowId} has no pending step for recovered correlationId {CorrelationId}; re- |
| | 2 | 347 | | await _builder.EnqueueWorkerAsync<IDurableFlowExecutor>(executor => executor.ExecuteAsync(flowId)).Configure |
| | 2 | 348 | | return; |
| | | 349 | | } |
| | | 350 | | |
| | 2 | 351 | | _logger.LogDebug("Durable flow {FlowId} checkpointed recovered correlationId {CorrelationId}; resuming.", flowId |
| | 2 | 352 | | await _builder.EnqueueWorkerAsync<IDurableFlowExecutor>(executor => executor.ExecuteAsync(flowId)).ConfigureAwai |
| | 2 | 353 | | } |
| | | 354 | | |
| | | 355 | | /// <inheritdoc /> |
| | | 356 | | public async Task FailAsync(string flowId, Exception exception) |
| | | 357 | | { |
| | 2 | 358 | | ArgumentException.ThrowIfNullOrWhiteSpace(flowId); |
| | 2 | 359 | | ArgumentNullException.ThrowIfNull(exception); |
| | | 360 | | |
| | 2 | 361 | | await using var scope = _scopeFactory.CreateAsyncScope(); |
| | 2 | 362 | | var store = scope.ServiceProvider.GetRequiredService<IFlowStateStore>(); |
| | | 363 | | |
| | 2 | 364 | | FlowState? updated = null; |
| | 2 | 365 | | var failedNow = false; |
| | 2 | 366 | | var found = await FlowStateConcurrency.MutateAsync( |
| | 2 | 367 | | store, |
| | 2 | 368 | | flowId, |
| | 2 | 369 | | _options.StateExpiry, |
| | 2 | 370 | | state => |
| | 2 | 371 | | { |
| | 2 | 372 | | updated = state; |
| | 2 | 373 | | failedNow = false; |
| | 2 | 374 | | if (state.Status != FlowRunStatus.Running) |
| | 2 | 375 | | return false; |
| | 2 | 376 | | |
| | 2 | 377 | | state.Status = FlowRunStatus.Failed; |
| | 2 | 378 | | state.LastMessage = exception.Message; |
| | 2 | 379 | | failedNow = true; |
| | 2 | 380 | | return true; |
| | 2 | 381 | | }).ConfigureAwait(false); |
| | | 382 | | |
| | 2 | 383 | | if (!found || updated is null) |
| | | 384 | | { |
| | 2 | 385 | | _logger.LogWarning("Durable flow {FlowId} cannot be failed: no state (unknown, expired, or unreadable).", fl |
| | 2 | 386 | | return; |
| | | 387 | | } |
| | | 388 | | |
| | 2 | 389 | | if (!failedNow) |
| | | 390 | | { |
| | 2 | 391 | | _logger.LogDebug("Durable flow {FlowId} is already {Status}; ignoring failure signal.", flowId, updated.Stat |
| | 2 | 392 | | await NotifyParentAsync(updated).ConfigureAwait(false); |
| | 2 | 393 | | return; |
| | | 394 | | } |
| | | 395 | | |
| | 2 | 396 | | await NotifyParentAsync(updated).ConfigureAwait(false); |
| | | 397 | | |
| | 2 | 398 | | _logger.LogWarning(exception, "Durable flow {FlowId} failed via lost-subscriber routing: {Message}", flowId, exc |
| | 2 | 399 | | } |
| | | 400 | | |
| | | 401 | | private async Task<bool> InvokeFlowAsync( |
| | | 402 | | IServiceProvider serviceProvider, |
| | | 403 | | IFlowStateStore store, |
| | | 404 | | FlowState state, |
| | | 405 | | FlowExecutionLease lease) |
| | | 406 | | { |
| | 3 | 407 | | var context = new DurableFlowContext( |
| | 3 | 408 | | state, |
| | 3 | 409 | | store, |
| | 3 | 410 | | _builder, |
| | 3 | 411 | | _propagation, |
| | 3 | 412 | | _options, |
| | 3 | 413 | | _subscriber, |
| | 3 | 414 | | _recoverableSubscriber, |
| | 3 | 415 | | _logger, |
| | 3 | 416 | | lease); |
| | | 417 | | |
| | | 418 | | // Statically-typed path for flows registered via WithDurableFlow<TFlow, TInput>(): no |
| | | 419 | | // type-name resolution, no MakeGenericType, no MethodInfo.Invoke — the path trimmed and |
| | | 420 | | // Native AOT apps rely on. |
| | 3 | 421 | | if (state.FlowTypeName is not null && _registrations.TryGetValue(state.FlowTypeName, out var registration)) |
| | | 422 | | { |
| | 1 | 423 | | var flow = ResolveFlowFromDi(serviceProvider, registration.FlowType); |
| | 1 | 424 | | var input = state.InputJson is null ? null : registration.DeserializeInput(state.InputJson); |
| | 1 | 425 | | await registration.ExecuteAsync(flow, context, input).ConfigureAwait(false); |
| | 1 | 426 | | await context.FlushProgressAsync().ConfigureAwait(false); |
| | 1 | 427 | | return context.IsSuspended; |
| | | 428 | | } |
| | | 429 | | |
| | 2 | 430 | | return await InvokeFlowByReflectionAsync(serviceProvider, state, context).ConfigureAwait(false); |
| | 3 | 431 | | } |
| | | 432 | | |
| | | 433 | | [UnconditionalSuppressMessage("Trimming", "IL2026", |
| | | 434 | | Justification = "Reflection fallback for flows not registered via WithDurableFlow<TFlow, TInput>(). In a trimmed |
| | | 435 | | "unregistered flow fails closed here with an actionable error telling the operator to register i |
| | | 436 | | "is silently misexecuted.")] |
| | | 437 | | [UnconditionalSuppressMessage("Trimming", "IL2075", |
| | | 438 | | Justification = "Same fallback contract: the flow type and IDurableFlow<TInput> instantiation exist whenever the |
| | | 439 | | "actually defines and starts the flow; otherwise resolution fails closed with guidance.")] |
| | | 440 | | [UnconditionalSuppressMessage("AOT", "IL3050", |
| | | 441 | | Justification = "MakeGenericType over the flow's input type re-materializes an interface instantiation the user' |
| | | 442 | | "class already implements statically; flows whose types were trimmed fail closed with guidance t |
| | | 443 | | "WithDurableFlow<TFlow, TInput>().")] |
| | | 444 | | private async Task<bool> InvokeFlowByReflectionAsync( |
| | | 445 | | IServiceProvider serviceProvider, |
| | | 446 | | FlowState state, |
| | | 447 | | DurableFlowContext context) |
| | | 448 | | { |
| | 2 | 449 | | var flowType = ResolveType(state.FlowTypeName, "flow"); |
| | 2 | 450 | | var inputType = ResolveType(state.InputTypeName, "input"); |
| | 2 | 451 | | var input = state.InputJson is null ? null : JsonSafety.SafeDeserialize(state.InputJson, inputType); |
| | | 452 | | |
| | 2 | 453 | | var contract = typeof(IDurableFlow<>).MakeGenericType(inputType); |
| | | 454 | | |
| | 2 | 455 | | var flow = ResolveFlowFromDi(serviceProvider, flowType); |
| | 2 | 456 | | if (!contract.IsInstanceOfType(flow)) |
| | | 457 | | { |
| | 2 | 458 | | throw new InvalidOperationException( |
| | 2 | 459 | | $"Durable flow type '{flowType.FullName}' does not implement IDurableFlow<{inputType.Name}> " + |
| | 2 | 460 | | "matching the persisted input type; the flow state was written by an incompatible flow definition."); |
| | | 461 | | } |
| | | 462 | | |
| | 2 | 463 | | var execute = contract.GetMethod(nameof(IDurableFlow<object>.ExecuteAsync))!; |
| | | 464 | | try |
| | | 465 | | { |
| | 2 | 466 | | await ((Task)execute.Invoke(flow, [context, input])!).ConfigureAwait(false); |
| | 2 | 467 | | await context.FlushProgressAsync().ConfigureAwait(false); |
| | 2 | 468 | | return context.IsSuspended; |
| | | 469 | | } |
| | 2 | 470 | | catch (System.Reflection.TargetInvocationException ex) when (ex.InnerException is not null) |
| | | 471 | | { |
| | | 472 | | // A synchronously-thrown flow exception arrives wrapped; unwrap so terminal |
| | | 473 | | // DurableFlowFailedException handling (and user-visible stack traces) see the real one. |
| | 2 | 474 | | System.Runtime.ExceptionServices.ExceptionDispatchInfo.Capture(ex.InnerException).Throw(); |
| | 0 | 475 | | throw; |
| | | 476 | | } |
| | 2 | 477 | | } |
| | | 478 | | |
| | | 479 | | private static object ResolveFlowFromDi(IServiceProvider serviceProvider, Type flowType) |
| | | 480 | | { |
| | | 481 | | try |
| | | 482 | | { |
| | 3 | 483 | | return serviceProvider.GetRequiredService(flowType); |
| | | 484 | | } |
| | 2 | 485 | | catch (InvalidOperationException ex) |
| | | 486 | | { |
| | 2 | 487 | | throw new InvalidOperationException( |
| | 2 | 488 | | $"Durable flow type '{flowType.FullName}' is not registered in DI. Register it with " + |
| | 2 | 489 | | $"WithDurableFlow<{flowType.Name}, TInput>() (or services.AddScoped<{flowType.Name}>()) so the flow can |
| | 2 | 490 | | "resolved on execute and resume.", ex); |
| | | 491 | | } |
| | 3 | 492 | | } |
| | | 493 | | |
| | | 494 | | private static Type ResolveType(string? fullName, string kind) |
| | | 495 | | { |
| | 2 | 496 | | if (string.IsNullOrWhiteSpace(fullName)) |
| | 2 | 497 | | throw new InvalidOperationException($"The persisted flow state carries no {kind} type name; it was written b |
| | | 498 | | |
| | 2 | 499 | | return ReflectionExtensions.ResolveServiceType(fullName) |
| | 2 | 500 | | ?? throw new InvalidOperationException( |
| | 2 | 501 | | $"Cannot resolve {kind} type '{fullName}'. For plugin/collectible-assembly scenarios register a resolver |
| | 2 | 502 | | $"via {nameof(AsyncResponseTypeResolution)}.{nameof(AsyncResponseTypeResolution.RegisterAssembly)}."); |
| | | 503 | | } |
| | | 504 | | |
| | | 505 | | private Task NotifyParentAsync(FlowState state) |
| | | 506 | | { |
| | 3 | 507 | | if (string.IsNullOrWhiteSpace(state.ParentFlowId)) |
| | 3 | 508 | | return Task.CompletedTask; |
| | | 509 | | |
| | | 510 | | // A suspended child cannot unblock its parent — the parent would only re-attach and go |
| | | 511 | | // back to waiting. The terminal transition after an operator un-suspends notifies then. |
| | 2 | 512 | | if (state.Status == FlowRunStatus.Suspended) |
| | 2 | 513 | | return Task.CompletedTask; |
| | | 514 | | |
| | 2 | 515 | | var parentFlowId = state.ParentFlowId; |
| | 2 | 516 | | _logger.LogInformation( |
| | 2 | 517 | | "Durable child flow {FlowId} reached {Status}; resuming parent flow {ParentFlowId} step '{ParentStepName}'." |
| | 2 | 518 | | state.FlowId, |
| | 2 | 519 | | state.Status, |
| | 2 | 520 | | parentFlowId, |
| | 2 | 521 | | state.ParentStepName); |
| | | 522 | | |
| | 2 | 523 | | return _builder.EnqueueWorkerAsync<IDurableFlowExecutor>(executor => executor.ExecuteAsync(parentFlowId)); |
| | | 524 | | } |
| | | 525 | | } |