| | | 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(string, System.Exception, string)"/> as their lost-subscriber callbacks — invoked by whichever |
| | | 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 | | /// Start target: the job <see cref="IDurableFlows.StartAsync{TFlow,TInput}"/> publishes. Creates |
| | | 28 | | /// the ledger from the serialized initial state the job carries when no ledger exists yet |
| | | 29 | | /// (insert-if-absent), then runs <see cref="ExecuteAsync"/>. The publish of this job — not the |
| | | 30 | | /// starter's own ledger write — is the start's commit point: a process that dies after the |
| | | 31 | | /// publish leaves a job whose execution creates the run, never a committed ledger that nothing |
| | | 32 | | /// will ever execute. An existing ledger for the same flow type and semantically identical |
| | | 33 | | /// input is executed as an idempotent re-start; one bound to different work is logged and the |
| | | 34 | | /// job dropped (the starter already reported the conflict to its caller). A start job whose |
| | | 35 | | /// carried state was created longer ago than <c>DurableFlowOptions.StateExpiry</c> and that |
| | | 36 | | /// finds no ledger is a replay of a run that has finished and expired: it is logged at Error |
| | | 37 | | /// and dropped rather than re-created, which would re-execute every completed step. |
| | | 38 | | /// </summary> |
| | | 39 | | Task CreateAndExecuteAsync(string flowId, string initialStateJson); |
| | | 40 | | |
| | | 41 | | /// <summary> |
| | | 42 | | /// Lost-subscriber resume target: re-enqueues <see cref="ExecuteAsync"/> on the worker |
| | | 43 | | /// transport (never runs the flow inline on a publisher's dispatch path). |
| | | 44 | | /// </summary> |
| | | 45 | | Task ResumeAsync(string flowId); |
| | | 46 | | |
| | | 47 | | /// <summary> |
| | | 48 | | /// Lost-subscriber success target: checkpoints the terminal payload into the matching pending |
| | | 49 | | /// step before re-enqueueing execution, so recovery does not wait for a consumed correlation id. |
| | | 50 | | /// </summary> |
| | | 51 | | Task RecoverAsync(string flowId, object payload, string correlationId); |
| | | 52 | | |
| | | 53 | | /// <summary>Lost-subscriber failure target: marks the run terminally <see cref="FlowRunStatus.Failed"/>.</summary> |
| | | 54 | | Task FailAsync(string flowId, Exception exception); |
| | | 55 | | |
| | | 56 | | /// <summary> |
| | | 57 | | /// Correlation-scoped lost-subscriber failure target: marks the run terminally |
| | | 58 | | /// <see cref="FlowRunStatus.Failed"/> only while a step is still pending on |
| | | 59 | | /// <paramref name="correlationId"/>. A failure for a correlation id the flow has since settled |
| | | 60 | | /// or superseded (a dead worker's registration outliving the replacement's, a late error for a |
| | | 61 | | /// step that already restarted fresh) is stale and is ignored — the same scoping |
| | | 62 | | /// <see cref="RecoverAsync"/> applies to the success target. |
| | | 63 | | /// </summary> |
| | | 64 | | Task FailAsync(string flowId, Exception exception, string correlationId); |
| | | 65 | | } |
| | | 66 | | |
| | | 67 | | /// <inheritdoc cref="IDurableFlowExecutor" /> |
| | | 68 | | internal sealed class DurableFlowExecutor : IDurableFlowExecutor |
| | | 69 | | { |
| | | 70 | | private readonly IServiceScopeFactory _scopeFactory; |
| | | 71 | | private readonly IAsyncResponseBuilder _builder; |
| | | 72 | | private readonly IAsyncResponseSubscriber _subscriber; |
| | | 73 | | private readonly IRecoverableAsyncResponseSubscriber? _recoverableSubscriber; |
| | | 74 | | private readonly AsyncResponseContextPropagation _propagation; |
| | | 75 | | private readonly DurableFlowOptions _options; |
| | | 76 | | private readonly ILogger<DurableFlowExecutor> _logger; |
| | | 77 | | private readonly TimeProvider _timeProvider; |
| | | 78 | | private readonly IDurableFlowExecutionObserver[] _observers; |
| | | 79 | | private readonly IWorkerTransport? _workerTransport; |
| | | 80 | | private readonly TimeSpan? _channelDefaultWaitTimeout; |
| | | 81 | | private readonly Dictionary<string, DurableFlowRegistration> _registrations; |
| | | 82 | | private readonly CancellationToken _hostStopping; |
| | | 83 | | |
| | | 84 | | /// <summary>Creates the flow executor.</summary> |
| | 1679 | 85 | | public DurableFlowExecutor( |
| | 1679 | 86 | | IServiceScopeFactory scopeFactory, |
| | 1679 | 87 | | IAsyncResponseBuilder builder, |
| | 1679 | 88 | | IAsyncResponseSubscriber subscriber, |
| | 1679 | 89 | | IRecoverableAsyncResponseSubscriber? recoverableSubscriber, |
| | 1679 | 90 | | AsyncResponseContextPropagation propagation, |
| | 1679 | 91 | | DurableFlowOptions options, |
| | 1679 | 92 | | ILogger<DurableFlowExecutor> logger, |
| | 1679 | 93 | | IEnumerable<DurableFlowRegistration>? registrations = null, |
| | 1679 | 94 | | Microsoft.Extensions.Hosting.IHostApplicationLifetime? hostLifetime = null, |
| | 1679 | 95 | | TimeProvider? timeProvider = null, |
| | 1679 | 96 | | IEnumerable<IDurableFlowExecutionObserver>? observers = null, |
| | 1679 | 97 | | IWorkerTransport? workerTransport = null, |
| | 1679 | 98 | | TimeSpan? channelDefaultWaitTimeout = null) |
| | | 99 | | { |
| | 1679 | 100 | | _workerTransport = workerTransport; |
| | 1679 | 101 | | _channelDefaultWaitTimeout = channelDefaultWaitTimeout; |
| | 1679 | 102 | | _scopeFactory = scopeFactory; |
| | 1679 | 103 | | _builder = builder; |
| | 1679 | 104 | | _subscriber = subscriber; |
| | 1679 | 105 | | _recoverableSubscriber = recoverableSubscriber; |
| | 1679 | 106 | | _propagation = propagation; |
| | 1679 | 107 | | _options = options; |
| | 1679 | 108 | | FlowStateConcurrency.ValidateOptions(_options); |
| | 1679 | 109 | | _logger = logger; |
| | 1679 | 110 | | _timeProvider = timeProvider ?? TimeProvider.System; |
| | 1679 | 111 | | _observers = observers?.ToArray() ?? []; |
| | 1679 | 112 | | _hostStopping = hostLifetime?.ApplicationStopping ?? CancellationToken.None; |
| | 1679 | 113 | | _registrations = new Dictionary<string, DurableFlowRegistration>(StringComparer.Ordinal); |
| | 6356 | 114 | | foreach (var registration in registrations ?? []) |
| | | 115 | | { |
| | | 116 | | // Last registration wins, matching DI's usual override semantics. |
| | 1499 | 117 | | _registrations[registration.FlowTypeFullName] = registration; |
| | | 118 | | } |
| | 1679 | 119 | | } |
| | | 120 | | |
| | | 121 | | /// <inheritdoc /> |
| | | 122 | | public async Task ExecuteAsync(string flowId) |
| | | 123 | | { |
| | 2025 | 124 | | ArgumentException.ThrowIfNullOrWhiteSpace(flowId); |
| | | 125 | | |
| | 2025 | 126 | | await using var scope = _scopeFactory.CreateAsyncScope(); |
| | 2025 | 127 | | var store = scope.ServiceProvider.GetRequiredService<IFlowStateStore>(); |
| | | 128 | | |
| | 2025 | 129 | | await using var lease = await AcquireExecutionLeaseWithRetryAsync(store, flowId).ConfigureAwait(false); |
| | 2009 | 130 | | if (lease is null) |
| | | 131 | | return; |
| | | 132 | | |
| | | 133 | | // A null load now means the run is genuinely gone (unknown, pruned, expired), which is the |
| | | 134 | | // one case where returning — and so acknowledging the wake-up — is correct. A ledger that |
| | | 135 | | // exists but cannot be read throws FlowStateUnreadableException instead and propagates to |
| | | 136 | | // the transport's retry/dead-letter policy, because acknowledging THAT abandons a Running |
| | | 137 | | // flow whose only remaining wake-up was this message. |
| | 1982 | 138 | | var state = await store.LoadAsync(flowId).ConfigureAwait(false); |
| | 1978 | 139 | | if (state is null) |
| | | 140 | | { |
| | 2 | 141 | | _logger.LogWarning("Durable flow {FlowId} has no state (unknown, pruned, or expired); nothing to execute.", |
| | 2 | 142 | | return; |
| | | 143 | | } |
| | | 144 | | |
| | 1976 | 145 | | if (state.Status != FlowRunStatus.Running) |
| | | 146 | | { |
| | 33 | 147 | | _logger.LogDebug("Durable flow {FlowId} is already {Status}; skipping execution.", flowId, state.Status); |
| | | 148 | | // Re-notify terminal runs on duplicate deliveries: the ORIGINAL delivery's |
| | | 149 | | // run-finished notification (below, outside the try/catch) may itself have thrown and |
| | | 150 | | // caused this redelivery — skipping here would lose the terminal event forever. |
| | | 151 | | // Observer delivery is at-least-once by contract; implementations tolerate duplicates. |
| | 33 | 152 | | if (state.Status is FlowRunStatus.Succeeded or FlowRunStatus.Failed) |
| | 31 | 153 | | await NotifyRunFinishedAsync(state).ConfigureAwait(false); |
| | 33 | 154 | | await NotifyParentAsync(state).ConfigureAwait(false); |
| | 33 | 155 | | return; |
| | | 156 | | } |
| | | 157 | | |
| | 1943 | 158 | | using var activity = AsyncResponseDiagnostics.StartActivity("asyncresponse.flow.execute"); |
| | 1943 | 159 | | activity?.SetTag("asyncresponse.flow_id", flowId); |
| | 1943 | 160 | | activity?.SetTag("asyncresponse.flow_type", state.FlowTypeName); |
| | | 161 | | |
| | 1943 | 162 | | state.Attempts++; |
| | 1943 | 163 | | await lease.SaveAsync(state, _options.StateExpiry).ConfigureAwait(false); |
| | | 164 | | |
| | | 165 | | // The run may be resumed by a different deployment than the one that started it: restore |
| | | 166 | | // the ambient context captured at start before any flow code runs. |
| | 1943 | 167 | | using var ambientScope = _propagation.Restore(state.Context); |
| | | 168 | | |
| | | 169 | | try |
| | | 170 | | { |
| | 1943 | 171 | | var suspended = await InvokeFlowAsync(scope.ServiceProvider, store, state, lease).ConfigureAwait(false); |
| | 954 | 172 | | if (suspended) |
| | | 173 | | { |
| | | 174 | | // The context persisted the suspended state BEFORE enqueueing the child; saving here |
| | | 175 | | // could overwrite newer checkpoints written by a parent re-execution the child has |
| | | 176 | | // already triggered on another worker. |
| | 2 | 177 | | _logger.LogDebug("Durable flow {FlowId} suspended: {Message}", flowId, state.LastMessage); |
| | 2 | 178 | | return; |
| | | 179 | | } |
| | | 180 | | |
| | 952 | 181 | | state.Status = FlowRunStatus.Succeeded; |
| | 952 | 182 | | state.LastMessage = "Flow completed."; |
| | 952 | 183 | | await lease.SaveAsync(state, _options.StateExpiry).ConfigureAwait(false); |
| | | 184 | | |
| | 952 | 185 | | _logger.LogInformation("Durable flow {FlowId} completed successfully (attempt {Attempts}).", flowId, state.A |
| | 952 | 186 | | } |
| | 174 | 187 | | catch (DurableFlowSuspendedException ex) |
| | | 188 | | { |
| | | 189 | | // Same as the IsSuspended return above: the suspended state is already persisted, and a |
| | | 190 | | // save here races the child-triggered parent re-execution. |
| | 174 | 191 | | _logger.LogDebug("Durable flow {FlowId} suspended: {Message}", flowId, ex.Message); |
| | 174 | 192 | | return; |
| | | 193 | | } |
| | 696 | 194 | | catch (DurableFlowFailedException ex) |
| | | 195 | | { |
| | | 196 | | // Terminal by declaration: mark failed and swallow so the transport acks the job. |
| | 696 | 197 | | state.Status = FlowRunStatus.Failed; |
| | 696 | 198 | | state.LastMessage = ex.Message; |
| | 696 | 199 | | await lease.SaveAsync(state, _options.StateExpiry, cause: ex).ConfigureAwait(false); |
| | | 200 | | |
| | 696 | 201 | | AsyncResponseDiagnostics.SetError(activity, ex); |
| | 696 | 202 | | _logger.LogWarning(ex, "Durable flow {FlowId} failed terminally: {Message}", flowId, ex.Message); |
| | 696 | 203 | | } |
| | 105 | 204 | | catch (Exception ex) when (lease.LostToken.IsCancellationRequested) |
| | | 205 | | { |
| | 10 | 206 | | AsyncResponseDiagnostics.SetError(activity, ex); |
| | 10 | 207 | | await NotifyRunAttemptFailedAsync(state).ConfigureAwait(false); |
| | 10 | 208 | | throw; |
| | 0 | 209 | | } |
| | 95 | 210 | | catch (Exception ex) |
| | | 211 | | { |
| | 95 | 212 | | state.LastMessage = ex.Message; |
| | 95 | 213 | | AsyncResponseDiagnostics.SetError(activity, ex); |
| | | 214 | | |
| | | 215 | | // Before the save: a rejected checkpoint below must not lose the notification (the |
| | | 216 | | // attempt ended either way, and any step it parked is no longer parked). |
| | 95 | 217 | | await NotifyRunAttemptFailedAsync(state).ConfigureAwait(false); |
| | | 218 | | |
| | 95 | 219 | | await lease.SaveAsync(state, _options.StateExpiry, cause: ex).ConfigureAwait(false); |
| | | 220 | | |
| | | 221 | | // Retriable: propagate so the worker transport redelivers the run with bounded |
| | | 222 | | // attempts and dead-letters it when they are exhausted — the "run is stuck" alarm. |
| | 87 | 223 | | throw; |
| | 0 | 224 | | } |
| | | 225 | | |
| | | 226 | | // The terminal outcome is persisted above; notify OUTSIDE the try/catch so a throwing |
| | | 227 | | // observer (throwing is the documented crash-injection contract) cannot re-enter those |
| | | 228 | | // catches and rewrite a Succeeded run as Failed — or overwrite a terminal ledger message |
| | | 229 | | // with a telemetry error — "the run's outcome is never at stake". A throw from here |
| | | 230 | | // propagates for redelivery; the replay sees the terminal status, RE-NOTIFIES (so the |
| | | 231 | | // event that just failed is not lost), and acks. |
| | 1648 | 232 | | await NotifyRunFinishedAsync(state).ConfigureAwait(false); |
| | 1648 | 233 | | await NotifyParentAsync(state).ConfigureAwait(false); |
| | 1886 | 234 | | } |
| | | 235 | | |
| | | 236 | | /// <summary> |
| | | 237 | | /// Acquires the execution lease for <paramref name="flowId"/>, retrying while the current |
| | | 238 | | /// holder's lease window elapses. Returns <c>null</c> when this delivery is safe to ack without |
| | | 239 | | /// executing (flow terminal/absent, or the lease is held by a demonstrably live worker). |
| | | 240 | | /// <para> |
| | | 241 | | /// A held lease alone is NOT proof this delivery is a duplicate: the holder may have died |
| | | 242 | | /// inside its unexpired lease window, and acking would drop the only wake-up the flow has — |
| | | 243 | | /// wake-ups would silently become at-most-once and the <see cref="FlowRunStatus.Running"/> run |
| | | 244 | | /// would strand. Neither is a lease that outlasts THIS host's lease window: the lease in the |
| | | 245 | | /// way may have been issued by another deployment with a longer |
| | | 246 | | /// <see cref="DurableFlowOptions.ExecutionLeaseDuration"/>, so a successor configured with a |
| | | 247 | | /// shorter one used to give up — and ack — before the dead holder's lease had even expired. |
| | | 248 | | /// </para> |
| | | 249 | | /// <para> |
| | | 250 | | /// Proof therefore comes from the store, not from elapsed local configuration |
| | | 251 | | /// (<see cref="IFlowStateStore.ObserveLeaseAsync"/>): a lease whose owner or expiry changes |
| | | 252 | | /// while this delivery waits was acquired or renewed by a live worker in the meantime; a lease |
| | | 253 | | /// that never changes is a dead holder's, and is waited out to its PERSISTED expiry. When the |
| | | 254 | | /// wait ends with neither proof nor the lease, the delivery is not acknowledged — |
| | | 255 | | /// <see cref="DurableFlowLeaseContendedException"/> hands it back to the transport. |
| | | 256 | | /// </para> |
| | | 257 | | /// <para> |
| | | 258 | | /// A live holder makes this delivery redundant only when the holder's OWN job is a different |
| | | 259 | | /// one: that job stays unacknowledged at the broker and is redelivered if the holder dies. A |
| | | 260 | | /// broker with an in-flight ceiling (Pub/Sub <c>MaxTotalAckExtension</c>, RabbitMQ |
| | | 261 | | /// <c>consumer_timeout</c>, the SQS 12-hour visibility cap, a Kafka rebalance) redelivers the |
| | | 262 | | /// holder's own job while its handler is still running, and THAT delivery is the last copy of |
| | | 263 | | /// the wake-up: acknowledging it leaves nothing to redeliver when the holder's process ends. |
| | | 264 | | /// The lease records the job that drives it (<see cref="FlowLeaseContention"/>), so such a |
| | | 265 | | /// delivery is recognised and never acknowledged as a duplicate — it is re-published as the |
| | | 266 | | /// same job, delayed past the lease, where the transport can delay, and otherwise kept with |
| | | 267 | | /// the transport. |
| | | 268 | | /// </para> |
| | | 269 | | /// </summary> |
| | | 270 | | private async Task<FlowExecutionLease?> AcquireExecutionLeaseWithRetryAsync(IFlowStateStore store, string flowId) |
| | | 271 | | { |
| | | 272 | | // The job driving this execution (null for a direct call, tag-less for a job written |
| | | 273 | | // before WorkerJobEnvelope.JobId existed): recorded with the lease on acquire, compared |
| | | 274 | | // with the lease in the way on contention. |
| | 2025 | 275 | | var ownJob = WorkerJobScope.Current; |
| | 2025 | 276 | | var ownJobTag = FlowLeaseContention.JobTag(ownJob?.JobId); |
| | | 277 | | |
| | 2025 | 278 | | var lease = await FlowStateConcurrency.TryAcquireExecutionLeaseAsync( |
| | 2025 | 279 | | store, |
| | 2025 | 280 | | flowId, |
| | 2025 | 281 | | _options, |
| | 2025 | 282 | | _logger, |
| | 2025 | 283 | | _timeProvider, |
| | 2025 | 284 | | jobTag: ownJobTag).ConfigureAwait(false); |
| | 2025 | 285 | | if (lease is not null) |
| | 1970 | 286 | | return lease; |
| | | 287 | | |
| | | 288 | | // This host's lease window bounds the wait only until the store says otherwise: the first |
| | | 289 | | // observation of the lease in the way moves the deadline to a full window past ITS expiry. |
| | | 290 | | // The 2s poll delay is capped by the renew interval so short test-sized leases still get polled. |
| | 55 | 291 | | var window = _options.ExecutionLeaseDuration + _options.ExecutionLeaseRenewInterval; |
| | 55 | 292 | | var startedWaitingUtc = _timeProvider.GetUtcNow().UtcDateTime; |
| | 55 | 293 | | var deadline = AddSaturating(startedWaitingUtc, window); |
| | | 294 | | |
| | | 295 | | // The store-driven extension below follows DATA this host does not control. A store clock |
| | | 296 | | // hours ahead of this one, or an expiry column read back shifted, used to move the deadline |
| | | 297 | | // as far out as the bad value said (saturating at DateTime.MaxValue, i.e. never): the |
| | | 298 | | // delivery then polled the store every pollDelay for good, pinning its worker slot, and |
| | | 299 | | // the contention exception whose message says "check for clock skew" was unreachable in |
| | | 300 | | // exactly the case it names. MaxLeaseContentionWait bounds the extension — never this |
| | | 301 | | // host's own window above, which is always waited. |
| | 55 | 302 | | var extensionCeiling = AddSaturating(startedWaitingUtc, _options.MaxLeaseContentionWait); |
| | 55 | 303 | | var extensionCapped = false; |
| | 55 | 304 | | var pollDelay = _options.ExecutionLeaseRenewInterval < TimeSpan.FromSeconds(2) |
| | 55 | 305 | | ? _options.ExecutionLeaseRenewInterval |
| | 55 | 306 | | : TimeSpan.FromSeconds(2); |
| | 55 | 307 | | FlowLeaseObservation? baseline = null; |
| | 55 | 308 | | var storeReportsLeases = true; |
| | 55 | 309 | | string? ownJobHolderLeaseId = null; |
| | | 310 | | |
| | 4116 | 311 | | while (true) |
| | | 312 | | { |
| | | 313 | | // Between attempts, look at the state itself: a terminal or absent flow needs no |
| | | 314 | | // execution, and reporting it accurately beats a misleading "already executing" log. |
| | 4171 | 315 | | var state = await store.LoadAsync(flowId).ConfigureAwait(false); |
| | 4171 | 316 | | if (state is null) |
| | | 317 | | { |
| | 4 | 318 | | _logger.LogWarning("Durable flow {FlowId} has no state (unknown, expired, or unreadable); nothing to exe |
| | 4 | 319 | | return null; |
| | | 320 | | } |
| | | 321 | | |
| | 4167 | 322 | | if (state.Status != FlowRunStatus.Running) |
| | | 323 | | { |
| | 5 | 324 | | _logger.LogDebug("Durable flow {FlowId} is already {Status}; skipping duplicate delivery.", flowId, stat |
| | | 325 | | // Same at-least-once re-notify as ExecuteAsync's terminal early return: the prior |
| | | 326 | | // delivery may have died in the run-finished notification itself. |
| | 5 | 327 | | if (state.Status is FlowRunStatus.Succeeded or FlowRunStatus.Failed) |
| | 5 | 328 | | await NotifyRunFinishedAsync(state).ConfigureAwait(false); |
| | 5 | 329 | | await NotifyParentAsync(state).ConfigureAwait(false); |
| | 5 | 330 | | return null; |
| | | 331 | | } |
| | | 332 | | |
| | 4162 | 333 | | lease = await FlowStateConcurrency.TryAcquireExecutionLeaseAsync( |
| | 4162 | 334 | | store, |
| | 4162 | 335 | | flowId, |
| | 4162 | 336 | | _options, |
| | 4162 | 337 | | _logger, |
| | 4162 | 338 | | _timeProvider, |
| | 4162 | 339 | | jobTag: ownJobTag).ConfigureAwait(false); |
| | 4162 | 340 | | if (lease is not null) |
| | 12 | 341 | | return lease; |
| | | 342 | | |
| | 4150 | 343 | | var observed = storeReportsLeases |
| | 4150 | 344 | | ? await store.ObserveLeaseAsync(flowId).ConfigureAwait(false) |
| | 4150 | 345 | | : null; |
| | 4150 | 346 | | if (observed is null) |
| | | 347 | | { |
| | 17 | 348 | | storeReportsLeases = false; |
| | | 349 | | } |
| | 4133 | 350 | | else if (observed.LeaseId is not null) |
| | | 351 | | { |
| | 4133 | 352 | | if (baseline is null) |
| | | 353 | | { |
| | 42 | 354 | | baseline = observed; |
| | 42 | 355 | | if (observed.ExpiresAtUtc is { } persistedExpiry) |
| | | 356 | | { |
| | | 357 | | // A dead holder's lease is acquirable once ITS expiry passes, whichever |
| | | 358 | | // deployment's lease duration issued it; the extra window absorbs clock |
| | | 359 | | // skew between this host and the store before the wait is declared stuck. |
| | 42 | 360 | | var persistedDeadline = AddSaturating(persistedExpiry, window); |
| | 42 | 361 | | if (persistedDeadline > extensionCeiling) |
| | | 362 | | { |
| | 18 | 363 | | persistedDeadline = extensionCeiling; |
| | 18 | 364 | | extensionCapped = true; |
| | | 365 | | } |
| | | 366 | | |
| | 42 | 367 | | if (persistedDeadline > deadline) |
| | 40 | 368 | | deadline = persistedDeadline; |
| | | 369 | | } |
| | | 370 | | } |
| | | 371 | | else |
| | | 372 | | { |
| | 4091 | 373 | | switch (FlowLeaseContention.Judge(baseline, observed, ownJobTag)) |
| | | 374 | | { |
| | | 375 | | case FlowLeaseContentionVerdict.AcknowledgeDuplicate: |
| | | 376 | | // Only a worker that acquired or renewed the lease AFTER this delivery |
| | | 377 | | // started waiting can have written that, and the job driving it is |
| | | 378 | | // not this one. The premise that makes the ack safe: the holder's OWN |
| | | 379 | | // job — a different job — is still unacknowledged at the broker, so |
| | | 380 | | // if that holder crashes later the broker redelivers it. The retry |
| | | 381 | | // loop exists to cover deliveries that arrive inside a DEAD holder's |
| | | 382 | | // unexpired lease window, which broker redelivery alone cannot cover. |
| | | 383 | | // Legacy caveat: when either side carries no job identity (a job or a |
| | | 384 | | // lease written before WorkerJobEnvelope.JobId existed, mid rolling |
| | | 385 | | // upgrade) the two cannot be told apart, and the evidence-based ack |
| | | 386 | | // applies as it did before. |
| | 14 | 387 | | _logger.LogDebug( |
| | 14 | 388 | | "Durable flow {FlowId} is executing on another live worker (its lease was {Evidence} whi |
| | 14 | 389 | | flowId, |
| | 14 | 390 | | string.Equals(observed.LeaseId, baseline.LeaseId, StringComparison.Ordinal) ? "renewed" |
| | 14 | 391 | | return null; |
| | | 392 | | |
| | | 393 | | case FlowLeaseContentionVerdict.HolderOwnJobRedelivered: |
| | | 394 | | // The live holder is executing THIS job: the broker handed it out a |
| | | 395 | | // second time while its handler was still running, which only happens |
| | | 396 | | // once an in-flight ceiling has lapsed. The first delivery can no |
| | | 397 | | // longer be settled, so this one is the last copy of the wake-up — |
| | | 398 | | // acknowledging it would leave nothing to redeliver when the holder's |
| | | 399 | | // process ends, and the run would stay Running forever. |
| | | 400 | | // MaxPublishDelay <= zero: the capability is unavailable in the current |
| | | 401 | | // configuration (an SQS FIFO worker queue) — same as not implementing it. |
| | 196 | 402 | | var redelay = _workerTransport is IDelayedWorkerTransport delayedTransport |
| | 196 | 403 | | && delayedTransport.MaxPublishDelay > TimeSpan.Zero |
| | 196 | 404 | | ? delayedTransport |
| | 196 | 405 | | : null; |
| | 196 | 406 | | if (ownJobHolderLeaseId is null) |
| | | 407 | | { |
| | 10 | 408 | | ownJobHolderLeaseId = observed.LeaseId; |
| | 10 | 409 | | AsyncResponseDiagnostics.RecordFlowOwnJobRedelivery(redelay is not null ? "redelayed" : |
| | 10 | 410 | | _logger.LogWarning( |
| | 10 | 411 | | "Durable flow {FlowId} wake-up is a redelivery of the job its live lease holder is s |
| | 10 | 412 | | "(Google Pub/Sub MaxTotalAckExtension, RabbitMQ consumer_timeout, the SQS 12-hour vi |
| | 10 | 413 | | flowId, |
| | 10 | 414 | | redelay is not null |
| | 10 | 415 | | ? "it is re-published as the same job, delayed past the holder's lease" |
| | 10 | 416 | | : "it waits for the lease and is otherwise handed back to the transport"); |
| | | 417 | | } |
| | | 418 | | |
| | 196 | 419 | | if (redelay is not null) |
| | | 420 | | { |
| | | 421 | | // Publish BEFORE the ack this return causes; a failed publish |
| | | 422 | | // propagates and the delivery stays with the transport. |
| | 6 | 423 | | await RepublishOwnJobPastLeaseAsync(redelay, ownJob!, observed, flowId).ConfigureAwait(f |
| | 4 | 424 | | return null; |
| | | 425 | | } |
| | | 426 | | |
| | | 427 | | // No delayed delivery: keep waiting on the deadline already set (a |
| | | 428 | | // live holder's renewals never extend it) — the lease freeing or the |
| | | 429 | | // ledger turning terminal resolves the wait, the deadline throws. |
| | | 430 | | break; |
| | | 431 | | } |
| | | 432 | | } |
| | | 433 | | } |
| | | 434 | | |
| | 4130 | 435 | | if (_timeProvider.GetUtcNow().UtcDateTime >= deadline) |
| | | 436 | | break; |
| | | 437 | | |
| | | 438 | | try |
| | | 439 | | { |
| | 4116 | 440 | | await Task.Delay(pollDelay, _timeProvider, _hostStopping).ConfigureAwait(false); |
| | 4116 | 441 | | } |
| | 0 | 442 | | catch (OperationCanceledException) |
| | | 443 | | { |
| | | 444 | | // Host shutdown must not leave this delivery parked in the poll — but acking it |
| | | 445 | | // would silently drop the flow's only wake-up. Propagate as cancellation so the |
| | | 446 | | // transport treats the job as not executed and redelivers it after restart. |
| | 0 | 447 | | throw new OperationCanceledException( |
| | 0 | 448 | | $"Host is stopping; durable flow '{flowId}' wake-up is abandoned for redelivery."); |
| | | 449 | | } |
| | 4116 | 450 | | } |
| | | 451 | | |
| | 14 | 452 | | if (ownJobHolderLeaseId is not null) |
| | | 453 | | { |
| | | 454 | | // A live holder WAS proven — and it is executing this delivery's own job, so the proof |
| | | 455 | | // is no licence to ack. The transport keeps the wake-up; its retry policy paces the |
| | | 456 | | // redeliveries and its dead-letter queue is the alarm if the holder outlives them. |
| | 4 | 457 | | throw new DurableFlowLeaseContendedException( |
| | 4 | 458 | | flowId, |
| | 4 | 459 | | $"the lease '{ownJobHolderLeaseId}' is held by a live execution of this same worker job: the broker rede |
| | 4 | 460 | | "(Google Pub/Sub MaxTotalAckExtension, RabbitMQ consumer_timeout, the SQS 12-hour visibility cap, or a K |
| | 4 | 461 | | "This delivery is the only copy of the wake-up the broker still has, so it is never acknowledged as a du |
| | 4 | 462 | | "keep a single in-process wait shorter than the broker's in-flight ceiling, or raise the ceiling"); |
| | | 463 | | } |
| | | 464 | | |
| | | 465 | | // No lease and no proof of a live holder. Acknowledging here is what stranded runs behind |
| | | 466 | | // a lease issued under a longer configuration; the transport keeps the wake-up instead. |
| | 10 | 467 | | throw new DurableFlowLeaseContendedException( |
| | 10 | 468 | | flowId, |
| | 10 | 469 | | !storeReportsLeases |
| | 10 | 470 | | ? $"the flow state store does not report leases ({nameof(IFlowStateStore)}.{nameof(IFlowStateStore.Obser |
| | 10 | 471 | | : baseline is null |
| | 10 | 472 | | ? $"the lease stayed unacquirable through this host's whole lease window of {window} although the st |
| | 10 | 473 | | : extensionCapped |
| | 10 | 474 | | ? $"the lease held by '{baseline.LeaseId}' (persisted expiry {baseline.ExpiresAtUtc:O}) neither |
| | 10 | 475 | | : $"the lease held by '{baseline.LeaseId}' (persisted expiry {baseline.ExpiresAtUtc:O}) neither |
| | 2009 | 476 | | } |
| | | 477 | | |
| | | 478 | | private static DateTime AddSaturating(DateTime instant, TimeSpan span) |
| | 158 | 479 | | => span > DateTime.MaxValue - instant ? DateTime.MaxValue : instant + span; |
| | | 480 | | |
| | | 481 | | /// <summary> |
| | | 482 | | /// Re-publishes <paramref name="job"/> — the job this delivery AND the live lease holder both |
| | | 483 | | /// carry — so that it comes back about when the holder's lease would lapse if the holder died |
| | | 484 | | /// now. The copy keeps the <see cref="WorkerJobEnvelope.JobId"/>: a fresh id would read as a |
| | | 485 | | /// different, redundant job at its next contention and be acknowledged on the holder's |
| | | 486 | | /// renewal, which is the loss this exists to prevent. The hop repeats at lease cadence while |
| | | 487 | | /// the holder lives; it finds a terminal ledger and acks once the holder finishes, or an |
| | | 488 | | /// expired lease it takes over once the holder dies. |
| | | 489 | | /// </summary> |
| | | 490 | | private async Task RepublishOwnJobPastLeaseAsync( |
| | | 491 | | IDelayedWorkerTransport transport, |
| | | 492 | | WorkerJobEnvelope job, |
| | | 493 | | FlowLeaseObservation holderLease, |
| | | 494 | | string flowId) |
| | | 495 | | { |
| | 6 | 496 | | var nowUtc = _timeProvider.GetUtcNow().UtcDateTime; |
| | | 497 | | |
| | | 498 | | // One renew interval past the persisted expiry, so a holder that is still alive has |
| | | 499 | | // visibly renewed by the time the hop lands. The expiry is store data this host does not |
| | | 500 | | // control (see MaxLeaseContentionWait above): bounded by that budget and by what the |
| | | 501 | | // transport can delay in one publish. Coming back early costs one more hop, never the run. |
| | 6 | 502 | | var delay = (holderLease.ExpiresAtUtc is { } expiresAtUtc && expiresAtUtc > nowUtc |
| | 6 | 503 | | ? expiresAtUtc - nowUtc |
| | 6 | 504 | | : TimeSpan.Zero) |
| | 6 | 505 | | + _options.ExecutionLeaseRenewInterval; |
| | 6 | 506 | | if (delay > _options.MaxLeaseContentionWait) |
| | 2 | 507 | | delay = _options.MaxLeaseContentionWait; |
| | 6 | 508 | | if (delay > transport.MaxPublishDelay) |
| | 0 | 509 | | delay = transport.MaxPublishDelay; |
| | | 510 | | |
| | 6 | 511 | | var hop = CopyForRedelay(job, AddSaturating(nowUtc, delay)); |
| | 6 | 512 | | await transport.PublishAsync(hop, delay).ConfigureAwait(false); |
| | | 513 | | |
| | 4 | 514 | | _logger.LogInformation( |
| | 4 | 515 | | "Durable flow {FlowId} wake-up re-published as the same job, due {NotBeforeUtc:O} ({Delay} from now, past th |
| | 4 | 516 | | flowId, |
| | 4 | 517 | | hop.NotBeforeUtc, |
| | 4 | 518 | | delay); |
| | 4 | 519 | | } |
| | | 520 | | |
| | | 521 | | /// <summary> |
| | | 522 | | /// The same job with a new due time. A copy, not the delivered instance: the transport still |
| | | 523 | | /// owns that one (the in-memory transport retries it as-is), and a due time stamped on it |
| | | 524 | | /// would outlive a failed publish. The stall counters are left behind on purpose — they |
| | | 525 | | /// belong to the due-time chain that ended with this delivery. |
| | | 526 | | /// </summary> |
| | 8 | 527 | | internal static WorkerJobEnvelope CopyForRedelay(WorkerJobEnvelope job, DateTime notBeforeUtc) => new() |
| | 8 | 528 | | { |
| | 8 | 529 | | SchemaVersion = job.SchemaVersion, |
| | 8 | 530 | | Call = job.Call, |
| | 8 | 531 | | CorrelationId = job.CorrelationId, |
| | 8 | 532 | | ReplyTarget = job.ReplyTarget, |
| | 8 | 533 | | Context = job.Context, |
| | 8 | 534 | | JobId = job.JobId, |
| | 8 | 535 | | NotBeforeUtc = notBeforeUtc |
| | 8 | 536 | | }; |
| | | 537 | | |
| | | 538 | | /// <inheritdoc /> |
| | | 539 | | public async Task CreateAndExecuteAsync(string flowId, string initialStateJson) |
| | | 540 | | { |
| | 1541 | 541 | | ArgumentException.ThrowIfNullOrWhiteSpace(flowId); |
| | 1541 | 542 | | ArgumentException.ThrowIfNullOrWhiteSpace(initialStateJson); |
| | | 543 | | |
| | | 544 | | // The carrier is the ledger wire format itself. A carrier this build cannot read is |
| | | 545 | | // deterministic: FlowStateUnreadableException propagates to the transport's retry and |
| | | 546 | | // dead-letter policy, which is the alarm — the same treatment an unreadable stored ledger |
| | | 547 | | // gets in ExecuteAsync, and for the same reason (acknowledging it would lose the start). |
| | 1541 | 548 | | var initial = FlowStateJson.Deserialize(initialStateJson, flowId); |
| | 1539 | 549 | | if (!string.Equals(initial.FlowId, flowId, StringComparison.Ordinal)) |
| | | 550 | | { |
| | 2 | 551 | | throw new InvalidOperationException( |
| | 2 | 552 | | $"The start job for durable flow '{flowId}' carries initial state for '{initial.FlowId}'; refusing to cr |
| | | 553 | | } |
| | | 554 | | |
| | 1537 | 555 | | await using (var scope = _scopeFactory.CreateAsyncScope()) |
| | | 556 | | { |
| | 1537 | 557 | | var store = scope.ServiceProvider.GetRequiredService<IFlowStateStore>(); |
| | | 558 | | |
| | | 559 | | // A start job outlives the run it started: a dead-letter replay, or a Kafka consumer |
| | | 560 | | // group rewound past it, delivers it again long after the run finished and its ledger |
| | | 561 | | // expired. Insert-if-absent then succeeds, and the "new" run re-executes every step — |
| | | 562 | | // and every side effect — of work that completed weeks ago. A start older than the |
| | | 563 | | // ledger lifetime with NO ledger behind it is that replay (or a start that never ran |
| | | 564 | | // and whose ledger would itself have expired by now, which an ExecuteAsync wake-up |
| | | 565 | | // would equally find gone): dropped, loudly. Checked only when the ledger is absent — |
| | | 566 | | // a live run past StateExpiry (a long park keeps its ledger through the retention |
| | | 567 | | // floor) still owns this job as its wake-up and takes the ordinary path below. |
| | 1537 | 568 | | if (StartedBeyondStateExpiry(initial, out var age) |
| | 1537 | 569 | | && await store.LoadAsync(flowId).ConfigureAwait(false) is null) |
| | | 570 | | { |
| | 2 | 571 | | _logger.LogError( |
| | 2 | 572 | | "Durable flow {FlowId} ({FlowType}) start job dropped: the start is {Age} old — past {StateExpiryOpt |
| | 2 | 573 | | flowId, initial.FlowTypeName, age, $"{nameof(DurableFlowOptions)}.{nameof(DurableFlowOptions.StateEx |
| | 2 | 574 | | return; |
| | | 575 | | } |
| | | 576 | | |
| | 1535 | 577 | | if (await FlowStateConcurrency.TryCreateAsync(store, flowId, initial, _options.StateExpiry).ConfigureAwait(f |
| | | 578 | | { |
| | | 579 | | // The starter died (or has not got there yet) between its publish and its own |
| | | 580 | | // create: the job is the durable record of the start, so the ledger comes from it. |
| | 663 | 581 | | _logger.LogInformation("Durable flow {FlowId} ({FlowType}) ledger created from its start job.", flowId, |
| | | 582 | | } |
| | | 583 | | else |
| | | 584 | | { |
| | 872 | 585 | | var existing = await store.LoadAsync(flowId).ConfigureAwait(false); |
| | 872 | 586 | | if (existing is null) |
| | | 587 | | { |
| | | 588 | | // Created and already expired or pruned between the two calls: genuinely gone, |
| | | 589 | | // the one case where acknowledging the start is right (ExecuteAsync's rule). |
| | 0 | 590 | | _logger.LogWarning("Durable flow {FlowId} exists but its ledger is expired or gone; nothing to execu |
| | 0 | 591 | | return; |
| | | 592 | | } |
| | | 593 | | |
| | 872 | 594 | | if (!FlowStateConcurrency.IsSameStart(existing, initial.FlowTypeName, initial.InputTypeName, initial.Inp |
| | | 595 | | { |
| | | 596 | | // The id was reused for different work. The starter that published this job |
| | | 597 | | // saw the same conflict on its own create and threw DurableFlowIdConflictException |
| | | 598 | | // to its caller; executing the EXISTING run here would wake a flow nobody asked |
| | | 599 | | // to wake, and creating a second one is impossible. Drop the job, loudly. |
| | 12 | 600 | | _logger.LogError( |
| | 12 | 601 | | "Durable flow {FlowId} start job dropped: the id is already bound to flow type {ExistingFlowType |
| | 12 | 602 | | flowId, existing.FlowTypeName, initial.FlowTypeName); |
| | 12 | 603 | | return; |
| | | 604 | | } |
| | | 605 | | |
| | | 606 | | // Same start, ledger already there (the starter's own create won, or this is a |
| | | 607 | | // redelivery / an idempotent re-start of a live run): fall through and execute it. |
| | | 608 | | } |
| | 1523 | 609 | | } |
| | | 610 | | |
| | 1523 | 611 | | await ExecuteAsync(flowId).ConfigureAwait(false); |
| | 1476 | 612 | | } |
| | | 613 | | |
| | | 614 | | /// <summary> |
| | | 615 | | /// Whether the start job's carried ledger was stamped longer ago than a ledger lives |
| | | 616 | | /// (<see cref="DurableFlowOptions.StateExpiry"/> — every save, the terminal one included, |
| | | 617 | | /// retains it for at least that long past <see cref="FlowState.CreatedAtUtc"/>). A carrier |
| | | 618 | | /// without the stamp is never judged: there is nothing to measure. |
| | | 619 | | /// </summary> |
| | | 620 | | private bool StartedBeyondStateExpiry(FlowState initial, out TimeSpan age) |
| | | 621 | | { |
| | 1537 | 622 | | age = TimeSpan.Zero; |
| | 1537 | 623 | | if (initial.CreatedAtUtc is not { } createdAt) |
| | 2 | 624 | | return false; |
| | | 625 | | |
| | 1535 | 626 | | var createdAtUtc = createdAt.Kind == DateTimeKind.Local ? createdAt.ToUniversalTime() : createdAt; |
| | 1535 | 627 | | age = _timeProvider.GetUtcNow().UtcDateTime - createdAtUtc; |
| | 1535 | 628 | | return age > _options.StateExpiry; |
| | | 629 | | } |
| | | 630 | | |
| | | 631 | | /// <inheritdoc /> |
| | | 632 | | public async Task ResumeAsync(string flowId) |
| | | 633 | | { |
| | 14 | 634 | | ArgumentException.ThrowIfNullOrWhiteSpace(flowId); |
| | | 635 | | |
| | 12 | 636 | | await using var scope = _scopeFactory.CreateAsyncScope(); |
| | 12 | 637 | | var store = scope.ServiceProvider.GetRequiredService<IFlowStateStore>(); |
| | | 638 | | |
| | 12 | 639 | | var state = await store.LoadAsync(flowId).ConfigureAwait(false); |
| | 12 | 640 | | if (state is null) |
| | | 641 | | { |
| | 6 | 642 | | _logger.LogWarning("Durable flow {FlowId} cannot resume: no state (unknown, expired, or unreadable).", flowI |
| | 6 | 643 | | return; |
| | | 644 | | } |
| | | 645 | | |
| | 6 | 646 | | if (state.Status != FlowRunStatus.Running) |
| | | 647 | | { |
| | 4 | 648 | | _logger.LogDebug("Durable flow {FlowId} is already {Status}; ignoring resume.", flowId, state.Status); |
| | 4 | 649 | | return; |
| | | 650 | | } |
| | | 651 | | |
| | 2 | 652 | | _logger.LogDebug("Durable flow {FlowId} resuming via worker transport.", flowId); |
| | 2 | 653 | | await _builder.EnqueueWorkerAsync<IDurableFlowExecutor>(executor => executor.ExecuteAsync(flowId)).ConfigureAwai |
| | 12 | 654 | | } |
| | | 655 | | |
| | | 656 | | /// <inheritdoc /> |
| | | 657 | | public async Task RecoverAsync(string flowId, object payload, string correlationId) |
| | | 658 | | { |
| | 42 | 659 | | ArgumentException.ThrowIfNullOrWhiteSpace(flowId); |
| | 40 | 660 | | ArgumentNullException.ThrowIfNull(payload); |
| | 38 | 661 | | ArgumentException.ThrowIfNullOrWhiteSpace(correlationId); |
| | | 662 | | |
| | 36 | 663 | | await using var scope = _scopeFactory.CreateAsyncScope(); |
| | 36 | 664 | | var store = scope.ServiceProvider.GetRequiredService<IFlowStateStore>(); |
| | 36 | 665 | | var checkpointed = false; |
| | 36 | 666 | | var running = false; |
| | 36 | 667 | | var lastStatus = FlowRunStatus.Running; |
| | 36 | 668 | | string? recoveredStep = null; |
| | | 669 | | |
| | 36 | 670 | | var found = await FlowStateConcurrency.MutateAsync( |
| | 36 | 671 | | store, |
| | 36 | 672 | | flowId, |
| | 36 | 673 | | _options.StateExpiry, |
| | 36 | 674 | | _timeProvider, |
| | 36 | 675 | | state => |
| | 36 | 676 | | { |
| | 34 | 677 | | checkpointed = false; |
| | 34 | 678 | | recoveredStep = null; |
| | 34 | 679 | | lastStatus = state.Status; |
| | 34 | 680 | | running = state.Status == FlowRunStatus.Running; |
| | 36 | 681 | | |
| | 36 | 682 | | // Suspended runs still CHECKPOINT the recovered terminal payload — the response |
| | 36 | 683 | | // exists nowhere else once this callback returns — but are never woken (see |
| | 36 | 684 | | // below): suspension means an operator took manual control, and ResumeAsync |
| | 36 | 685 | | // continues from the checkpoint instead of re-running the remote step. |
| | 34 | 686 | | var checkpointable = running || state.Status == FlowRunStatus.Suspended; |
| | 34 | 687 | | if (!checkpointable || state.Steps is null) |
| | 6 | 688 | | return false; |
| | 36 | 689 | | |
| | 28 | 690 | | var pending = state.Steps.FirstOrDefault(pair => |
| | 58 | 691 | | string.Equals(pair.Value.PendingCorrelationId, correlationId, StringComparison.Ordinal)); |
| | 28 | 692 | | if (pending.Value is null) |
| | 8 | 693 | | return false; |
| | 36 | 694 | | |
| | 20 | 695 | | pending.Value.Completed = true; |
| | 20 | 696 | | pending.Value.ResultJson = SerializeRecoveredResult(payload, pending.Value.PendingPayloadTypeFullName); |
| | 20 | 697 | | pending.Value.PendingCorrelationId = null; |
| | 20 | 698 | | pending.Value.PendingPayloadTypeFullName = null; |
| | 20 | 699 | | pending.Value.Faulted = false; |
| | 20 | 700 | | pending.Value.Message = "Terminal response recovered after subscriber loss."; |
| | 20 | 701 | | pending.Value.CompletedAtUtc = _timeProvider.GetUtcNow().UtcDateTime; |
| | 20 | 702 | | state.LastMessage = $"Step '{pending.Key}' recovered after subscriber loss."; |
| | 20 | 703 | | checkpointed = true; |
| | 20 | 704 | | recoveredStep = pending.Key; |
| | 20 | 705 | | return true; |
| | 36 | 706 | | }).ConfigureAwait(false); |
| | | 707 | | |
| | 36 | 708 | | if (!found) |
| | | 709 | | { |
| | 2 | 710 | | _logger.LogWarning("Durable flow {FlowId} cannot recover response {CorrelationId}: no state found.", flowId, |
| | 2 | 711 | | return; |
| | | 712 | | } |
| | | 713 | | |
| | 34 | 714 | | if (!checkpointed) |
| | | 715 | | { |
| | 14 | 716 | | if (!running) |
| | | 717 | | { |
| | 4 | 718 | | _logger.LogDebug("Durable flow {FlowId} is {Status}; ignoring recovered correlationId {CorrelationId}.", |
| | 4 | 719 | | return; |
| | | 720 | | } |
| | | 721 | | |
| | | 722 | | // Still Running with no matching pending step: a previous delivery may have crashed |
| | | 723 | | // between checkpointing this response and enqueueing the run, making this redelivery |
| | | 724 | | // the only remaining wake-up. Re-enqueue instead of dropping — it is idempotent, and |
| | | 725 | | // worst case the job finds a live holder's lease and acks as a duplicate. |
| | 10 | 726 | | _logger.LogDebug("Durable flow {FlowId} has no pending step for recovered correlationId {CorrelationId}; re- |
| | 10 | 727 | | await _builder.EnqueueWorkerAsync<IDurableFlowExecutor>(executor => executor.ExecuteAsync(flowId)).Configure |
| | 10 | 728 | | return; |
| | | 729 | | } |
| | | 730 | | |
| | | 731 | | // The completion recorded here is the ONLY chance observers get to see this step finish: |
| | | 732 | | // the replayed execution short-circuits the now-memoized step without notifying. Notified |
| | | 733 | | // before the wake-up is enqueued so observers (e.g. the Testing probe's step waiters) see |
| | | 734 | | // the completion before the resumed run races past it. Best-effort by necessity: once the |
| | | 735 | | // checkpoint settles, the pending correlation id is gone, and a redelivered RecoverAsync |
| | | 736 | | // can no longer tell which step this response completed — an observer that throws here |
| | | 737 | | // loses the event (unlike run-finished, which re-derives from the persisted status). |
| | 20 | 738 | | await NotifyStepCompletedAsync(flowId, recoveredStep!, correlationId).ConfigureAwait(false); |
| | | 739 | | |
| | 20 | 740 | | if (!running) |
| | | 741 | | { |
| | 6 | 742 | | _logger.LogInformation( |
| | 6 | 743 | | "Durable flow {FlowId} checkpointed recovered correlationId {CorrelationId} while Suspended; not waking |
| | 6 | 744 | | flowId, correlationId); |
| | 6 | 745 | | return; |
| | | 746 | | } |
| | | 747 | | |
| | 14 | 748 | | _logger.LogDebug("Durable flow {FlowId} checkpointed recovered correlationId {CorrelationId}; resuming.", flowId |
| | 14 | 749 | | await _builder.EnqueueWorkerAsync<IDurableFlowExecutor>(executor => executor.ExecuteAsync(flowId)).ConfigureAwai |
| | 32 | 750 | | } |
| | | 751 | | |
| | | 752 | | private async Task NotifyStepCompletedAsync(string flowId, string stepName, string correlationId) |
| | | 753 | | { |
| | 20 | 754 | | if (_observers.Length == 0) |
| | 14 | 755 | | return; |
| | | 756 | | |
| | 6 | 757 | | var stepEvent = new DurableFlowStepEvent(flowId, stepName, DurableFlowStepKind.Awaited, correlationId, WakeAtUtc |
| | 28 | 758 | | foreach (var observer in _observers) |
| | 8 | 759 | | await observer.OnStepCompletedAsync(stepEvent).ConfigureAwait(false); |
| | 20 | 760 | | } |
| | | 761 | | |
| | | 762 | | /// <summary> |
| | | 763 | | /// Serializes a recovered terminal payload for the step checkpoint AS THE STEP'S DECLARED |
| | | 764 | | /// response type (recorded at await time): replay deserializes the checkpoint as that declared |
| | | 765 | | /// type, and the recovered payload's runtime type may be a <c>[JsonPolymorphic]</c> derived |
| | | 766 | | /// whose runtime-type serialization omits the discriminator — breaking every replay against an |
| | | 767 | | /// abstract declared base, or silently truncating against a concrete one. Falls back to the |
| | | 768 | | /// runtime type when the recorded name is absent (ledgers written before it existed) or no |
| | | 769 | | /// longer resolves to a compatible type. |
| | | 770 | | /// </summary> |
| | | 771 | | private static string SerializeRecoveredResult(object payload, string? declaredTypeFullName) |
| | | 772 | | { |
| | 20 | 773 | | var declaredType = declaredTypeFullName is null |
| | 20 | 774 | | ? null |
| | 20 | 775 | | : PayloadRecoveryClassifier.ResolvePayloadType(declaredTypeFullName); |
| | | 776 | | |
| | 20 | 777 | | return declaredType is not null && declaredType.IsInstanceOfType(payload) |
| | 20 | 778 | | ? AsyncResponseJson.Serialize(payload, declaredType) |
| | 20 | 779 | | : AsyncResponseJson.Serialize(payload, payload.GetType()); |
| | | 780 | | } |
| | | 781 | | |
| | | 782 | | /// <inheritdoc /> |
| | | 783 | | public Task FailAsync(string flowId, Exception exception) |
| | 14 | 784 | | => FailCoreAsync(flowId, exception, correlationId: null); |
| | | 785 | | |
| | | 786 | | public Task FailAsync(string flowId, Exception exception, string correlationId) |
| | | 787 | | { |
| | 6 | 788 | | ArgumentException.ThrowIfNullOrWhiteSpace(correlationId); |
| | 6 | 789 | | return FailCoreAsync(flowId, exception, correlationId); |
| | | 790 | | } |
| | | 791 | | |
| | | 792 | | private async Task FailCoreAsync(string flowId, Exception exception, string? correlationId) |
| | | 793 | | { |
| | 20 | 794 | | ArgumentException.ThrowIfNullOrWhiteSpace(flowId); |
| | 18 | 795 | | ArgumentNullException.ThrowIfNull(exception); |
| | | 796 | | |
| | 16 | 797 | | await using var scope = _scopeFactory.CreateAsyncScope(); |
| | 16 | 798 | | var store = scope.ServiceProvider.GetRequiredService<IFlowStateStore>(); |
| | | 799 | | |
| | 16 | 800 | | FlowState? updated = null; |
| | 16 | 801 | | var failedNow = false; |
| | 16 | 802 | | var stale = false; |
| | 16 | 803 | | var found = await FlowStateConcurrency.MutateAsync( |
| | 16 | 804 | | store, |
| | 16 | 805 | | flowId, |
| | 16 | 806 | | _options.StateExpiry, |
| | 16 | 807 | | _timeProvider, |
| | 16 | 808 | | state => |
| | 16 | 809 | | { |
| | 14 | 810 | | updated = state; |
| | 14 | 811 | | failedNow = false; |
| | 14 | 812 | | stale = false; |
| | 14 | 813 | | if (state.Status != FlowRunStatus.Running) |
| | 2 | 814 | | return false; |
| | 16 | 815 | | |
| | 16 | 816 | | // A correlation-scoped failure only counts against the step still pending on |
| | 16 | 817 | | // that id (RecoverAsync parity): a dead worker's registration outlives the |
| | 16 | 818 | | // replacement's, so a late error for a superseded or already-settled correlation |
| | 16 | 819 | | // id must not fail a run that is live on another one. |
| | 12 | 820 | | if (correlationId is not null |
| | 12 | 821 | | && (state.Steps is null |
| | 18 | 822 | | || !state.Steps.Values.Any(step => string.Equals(step.PendingCorrelationId, correlationId, Strin |
| | 16 | 823 | | { |
| | 4 | 824 | | stale = true; |
| | 4 | 825 | | return false; |
| | 16 | 826 | | } |
| | 16 | 827 | | |
| | 8 | 828 | | state.Status = FlowRunStatus.Failed; |
| | 8 | 829 | | state.LastMessage = exception.Message; |
| | 8 | 830 | | failedNow = true; |
| | 8 | 831 | | return true; |
| | 16 | 832 | | }).ConfigureAwait(false); |
| | | 833 | | |
| | 16 | 834 | | if (!found || updated is null) |
| | | 835 | | { |
| | 2 | 836 | | _logger.LogWarning("Durable flow {FlowId} cannot be failed: no state (unknown, expired, or unreadable).", fl |
| | 2 | 837 | | return; |
| | | 838 | | } |
| | | 839 | | |
| | 14 | 840 | | if (stale) |
| | | 841 | | { |
| | 4 | 842 | | _logger.LogDebug("Durable flow {FlowId} has no step pending on correlationId {CorrelationId}; ignoring stale |
| | 4 | 843 | | return; |
| | | 844 | | } |
| | | 845 | | |
| | 10 | 846 | | if (!failedNow) |
| | | 847 | | { |
| | 2 | 848 | | _logger.LogDebug("Durable flow {FlowId} is already {Status}; ignoring failure signal.", flowId, updated.Stat |
| | | 849 | | // At-least-once re-notify, as in ExecuteAsync: the prior delivery of this failure |
| | | 850 | | // signal may have marked the run and then died in its own run-finished notification. |
| | 2 | 851 | | if (updated.Status is FlowRunStatus.Succeeded or FlowRunStatus.Failed) |
| | 2 | 852 | | await NotifyRunFinishedAsync(updated).ConfigureAwait(false); |
| | 2 | 853 | | await NotifyParentAsync(updated).ConfigureAwait(false); |
| | 2 | 854 | | return; |
| | | 855 | | } |
| | | 856 | | |
| | | 857 | | // Run-finished observers BEFORE the parent wake-up, matching ExecuteAsync and the |
| | | 858 | | // already-terminal branch above: enqueueing the parent first would let it resume and |
| | | 859 | | // finish before this run's terminal event is recorded, and an observer throw after the |
| | | 860 | | // parent was already notified would re-enqueue the parent a second time on redelivery. |
| | 8 | 861 | | await NotifyRunFinishedAsync(updated).ConfigureAwait(false); |
| | 6 | 862 | | await NotifyParentAsync(updated).ConfigureAwait(false); |
| | | 863 | | |
| | 6 | 864 | | _logger.LogWarning(exception, "Durable flow {FlowId} failed via lost-subscriber routing: {Message}", flowId, exc |
| | 14 | 865 | | } |
| | | 866 | | |
| | | 867 | | private async Task<bool> InvokeFlowAsync( |
| | | 868 | | IServiceProvider serviceProvider, |
| | | 869 | | IFlowStateStore store, |
| | | 870 | | FlowState state, |
| | | 871 | | FlowExecutionLease lease) |
| | | 872 | | { |
| | 1943 | 873 | | var context = new DurableFlowContext( |
| | 1943 | 874 | | state, |
| | 1943 | 875 | | store, |
| | 1943 | 876 | | _builder, |
| | 1943 | 877 | | _propagation, |
| | 1943 | 878 | | _options, |
| | 1943 | 879 | | _subscriber, |
| | 1943 | 880 | | _recoverableSubscriber, |
| | 1943 | 881 | | _logger, |
| | 1943 | 882 | | lease, |
| | 1943 | 883 | | _timeProvider, |
| | 1943 | 884 | | _observers, |
| | 1943 | 885 | | _workerTransport, |
| | 1943 | 886 | | _channelDefaultWaitTimeout); |
| | | 887 | | |
| | | 888 | | // Statically-typed path for flows registered via WithDurableFlow<TFlow, TInput>(): no |
| | | 889 | | // type-name resolution, no MakeGenericType, no MethodInfo.Invoke — the path trimmed and |
| | | 890 | | // Native AOT apps rely on. |
| | 1943 | 891 | | if (state.FlowTypeName is not null && _registrations.TryGetValue(state.FlowTypeName, out var registration)) |
| | | 892 | | { |
| | | 893 | | // Fail closed exactly like the reflection fallback: a run persisted with a different |
| | | 894 | | // input type than the registration's TInput would otherwise silently parse the old |
| | | 895 | | // payload as the new type — renamed or added members become defaults — and execute |
| | | 896 | | // the remaining steps against wrong input. Null tolerated: ledgers written before the |
| | | 897 | | // stamp existed carry no input type name. |
| | 1771 | 898 | | if (state.InputTypeName is not null |
| | 1771 | 899 | | && !string.Equals(state.InputTypeName, registration.InputTypeFullName, StringComparison.Ordinal)) |
| | | 900 | | { |
| | 2 | 901 | | throw new InvalidOperationException( |
| | 2 | 902 | | $"Durable flow type '{state.FlowTypeName}' is registered with input type '{registration.InputTypeFul |
| | 2 | 903 | | $"but the persisted run carries input type '{state.InputTypeName}'; the flow state was written by an |
| | 2 | 904 | | "incompatible flow definition."); |
| | | 905 | | } |
| | | 906 | | |
| | 1769 | 907 | | var flow = ResolveFlowFromDi(serviceProvider, registration.FlowType); |
| | 1769 | 908 | | var input = state.InputJson is null ? null : registration.DeserializeInput(state.InputJson); |
| | 1769 | 909 | | await registration.ExecuteAsync(flow, context, input).ConfigureAwait(false); |
| | 862 | 910 | | await context.FlushProgressAsync().ConfigureAwait(false); |
| | 862 | 911 | | return context.IsSuspended; |
| | | 912 | | } |
| | | 913 | | |
| | 172 | 914 | | return await InvokeFlowByReflectionAsync(serviceProvider, state, context).ConfigureAwait(false); |
| | 954 | 915 | | } |
| | | 916 | | |
| | | 917 | | [UnconditionalSuppressMessage("Trimming", "IL2026", |
| | | 918 | | Justification = "Reflection fallback for flows not registered via WithDurableFlow<TFlow, TInput>(). In a trimmed |
| | | 919 | | "unregistered flow fails closed here with an actionable error telling the operator to register i |
| | | 920 | | "is silently misexecuted.")] |
| | | 921 | | [UnconditionalSuppressMessage("Trimming", "IL2075", |
| | | 922 | | Justification = "Same fallback contract: the flow type and IDurableFlow<TInput> instantiation exist whenever the |
| | | 923 | | "actually defines and starts the flow; otherwise resolution fails closed with guidance.")] |
| | | 924 | | [UnconditionalSuppressMessage("AOT", "IL3050", |
| | | 925 | | Justification = "MakeGenericType over the flow's input type re-materializes an interface instantiation the user' |
| | | 926 | | "class already implements statically; flows whose types were trimmed fail closed with guidance t |
| | | 927 | | "WithDurableFlow<TFlow, TInput>().")] |
| | | 928 | | private async Task<bool> InvokeFlowByReflectionAsync( |
| | | 929 | | IServiceProvider serviceProvider, |
| | | 930 | | FlowState state, |
| | | 931 | | DurableFlowContext context) |
| | | 932 | | { |
| | 172 | 933 | | var flowType = ResolveType(state.FlowTypeName, "flow"); |
| | 168 | 934 | | var inputType = ResolveType(state.InputTypeName, "input"); |
| | | 935 | | |
| | 166 | 936 | | var contract = typeof(IDurableFlow<>).MakeGenericType(inputType); |
| | | 937 | | |
| | 166 | 938 | | var flow = ResolveFlowFromDi(serviceProvider, flowType); |
| | 164 | 939 | | if (!contract.IsInstanceOfType(flow)) |
| | | 940 | | { |
| | 4 | 941 | | throw new InvalidOperationException( |
| | 4 | 942 | | $"Durable flow type '{flowType.FullName}' does not implement IDurableFlow<{inputType.Name}> " + |
| | 4 | 943 | | "matching the persisted input type; the flow state was written by an incompatible flow definition."); |
| | | 944 | | } |
| | | 945 | | |
| | | 946 | | // Deserialize only AFTER the contract check above has passed: the ledger's InputTypeName is |
| | | 947 | | // attacker-controlled to anyone who can write the flow store, and STJ construction runs |
| | | 948 | | // setters/converters/[JsonConstructor] on whatever type it materializes. Requiring a |
| | | 949 | | // DI-registered flow that implements IDurableFlow<inputType> first bounds the constructible |
| | | 950 | | // set to input types the application actually declared, instead of any loadable CLR type. |
| | 160 | 951 | | var input = state.InputJson is null ? null : JsonSafety.SafeDeserialize(state.InputJson, inputType); |
| | | 952 | | |
| | 160 | 953 | | var execute = contract.GetMethod(nameof(IDurableFlow<object>.ExecuteAsync))!; |
| | | 954 | | try |
| | | 955 | | { |
| | 160 | 956 | | await ((Task)execute.Invoke(flow, [context, input])!).ConfigureAwait(false); |
| | 92 | 957 | | await context.FlushProgressAsync().ConfigureAwait(false); |
| | 92 | 958 | | return context.IsSuspended; |
| | | 959 | | } |
| | 6 | 960 | | catch (System.Reflection.TargetInvocationException ex) when (ex.InnerException is not null) |
| | | 961 | | { |
| | | 962 | | // A synchronously-thrown flow exception arrives wrapped; unwrap so terminal |
| | | 963 | | // DurableFlowFailedException handling (and user-visible stack traces) see the real one. |
| | 6 | 964 | | System.Runtime.ExceptionServices.ExceptionDispatchInfo.Capture(ex.InnerException).Throw(); |
| | 0 | 965 | | throw; |
| | | 966 | | } |
| | 92 | 967 | | } |
| | | 968 | | |
| | | 969 | | private static object ResolveFlowFromDi(IServiceProvider serviceProvider, Type flowType) |
| | | 970 | | { |
| | | 971 | | try |
| | | 972 | | { |
| | 1935 | 973 | | return serviceProvider.GetRequiredService(flowType); |
| | | 974 | | } |
| | 2 | 975 | | catch (InvalidOperationException ex) |
| | | 976 | | { |
| | 2 | 977 | | throw new InvalidOperationException( |
| | 2 | 978 | | $"Durable flow type '{flowType.FullName}' is not registered in DI. Register it with " + |
| | 2 | 979 | | $"WithDurableFlow<{flowType.Name}, TInput>() (or services.AddScoped<{flowType.Name}>()) so the flow can |
| | 2 | 980 | | "resolved on execute and resume.", ex); |
| | | 981 | | } |
| | 1933 | 982 | | } |
| | | 983 | | |
| | | 984 | | private static Type ResolveType(string? fullName, string kind) |
| | | 985 | | { |
| | 340 | 986 | | if (string.IsNullOrWhiteSpace(fullName)) |
| | 4 | 987 | | throw new InvalidOperationException($"The persisted flow state carries no {kind} type name; it was written b |
| | | 988 | | |
| | 336 | 989 | | return ReflectionExtensions.ResolveServiceType(fullName) |
| | 336 | 990 | | // The name is store data: rendered through the diagnostics helper so an unresolvable |
| | 336 | 991 | | // one cannot copy megabytes of store-written text, or its raw line breaks, into this |
| | 336 | 992 | | // message and from there into a log on every delivery. |
| | 336 | 993 | | ?? throw new InvalidOperationException( |
| | 336 | 994 | | $"Cannot resolve {kind} type '{AsyncResponseTypeResolution.DescribeForDiagnostics(fullName)}'. For plugi |
| | 336 | 995 | | $"via {nameof(AsyncResponseTypeResolution)}.{nameof(AsyncResponseTypeResolution.RegisterAssembly)}."); |
| | | 996 | | } |
| | | 997 | | |
| | | 998 | | /// <summary> |
| | | 999 | | /// Observer hook for terminal transitions. Fires AFTER the terminal save: an observer that |
| | | 1000 | | /// throws here (crash injection) fails a delivery whose run is already terminal, so the |
| | | 1001 | | /// redelivered execution re-notifies and acks — the run's outcome is never at stake, and the |
| | | 1002 | | /// terminal event is delivered at least once. |
| | | 1003 | | /// </summary> |
| | | 1004 | | /// <summary> |
| | | 1005 | | /// Best-effort attempt-failure notification (see |
| | | 1006 | | /// <see cref="IDurableFlowExecutionObserver.OnRunAttemptFailedAsync"/>): the attempt's own |
| | | 1007 | | /// exception is already propagating for redelivery, so observer throws are swallowed instead |
| | | 1008 | | /// of masking it. |
| | | 1009 | | /// </summary> |
| | | 1010 | | private async Task NotifyRunAttemptFailedAsync(FlowState state) |
| | | 1011 | | { |
| | 105 | 1012 | | if (_observers.Length == 0) |
| | 47 | 1013 | | return; |
| | | 1014 | | |
| | 58 | 1015 | | var runEvent = new DurableFlowRunEvent(state.FlowId!, state.Status, state.LastMessage); |
| | 324 | 1016 | | foreach (var observer in _observers) |
| | | 1017 | | { |
| | | 1018 | | try |
| | | 1019 | | { |
| | 104 | 1020 | | await observer.OnRunAttemptFailedAsync(runEvent).ConfigureAwait(false); |
| | 102 | 1021 | | } |
| | 2 | 1022 | | catch (Exception ex) |
| | | 1023 | | { |
| | 2 | 1024 | | _logger.LogWarning( |
| | 2 | 1025 | | ex, |
| | 2 | 1026 | | "A durable-flow execution observer threw in OnRunAttemptFailedAsync for {FlowId}; ignoring so the at |
| | 2 | 1027 | | state.FlowId); |
| | 2 | 1028 | | } |
| | | 1029 | | } |
| | 105 | 1030 | | } |
| | | 1031 | | |
| | | 1032 | | private async Task NotifyRunFinishedAsync(FlowState state) |
| | | 1033 | | { |
| | 1694 | 1034 | | if (_observers.Length == 0) |
| | 1468 | 1035 | | return; |
| | | 1036 | | |
| | 226 | 1037 | | var runEvent = new DurableFlowRunEvent(state.FlowId!, state.Status, state.LastMessage); |
| | 1334 | 1038 | | foreach (var observer in _observers) |
| | 442 | 1039 | | await observer.OnRunFinishedAsync(runEvent).ConfigureAwait(false); |
| | 1692 | 1040 | | } |
| | | 1041 | | |
| | | 1042 | | private Task NotifyParentAsync(FlowState state) |
| | | 1043 | | { |
| | 1694 | 1044 | | if (string.IsNullOrWhiteSpace(state.ParentFlowId)) |
| | 1580 | 1045 | | return Task.CompletedTask; |
| | | 1046 | | |
| | | 1047 | | // A suspended child cannot unblock its parent — the parent would only re-attach and go |
| | | 1048 | | // back to waiting. The terminal transition after an operator un-suspends notifies then. |
| | 114 | 1049 | | if (state.Status == FlowRunStatus.Suspended) |
| | 2 | 1050 | | return Task.CompletedTask; |
| | | 1051 | | |
| | 112 | 1052 | | var parentFlowId = state.ParentFlowId; |
| | 112 | 1053 | | _logger.LogInformation( |
| | 112 | 1054 | | "Durable child flow {FlowId} reached {Status}; resuming parent flow {ParentFlowId} step '{ParentStepName}'." |
| | 112 | 1055 | | state.FlowId, |
| | 112 | 1056 | | state.Status, |
| | 112 | 1057 | | parentFlowId, |
| | 112 | 1058 | | state.ParentStepName); |
| | | 1059 | | |
| | 112 | 1060 | | return _builder.EnqueueWorkerAsync<IDurableFlowExecutor>(executor => executor.ExecuteAsync(parentFlowId)); |
| | | 1061 | | } |
| | | 1062 | | } |