| | | 1 | | using Microsoft.Extensions.Logging; |
| | | 2 | | using System.Diagnostics.CodeAnalysis; |
| | | 3 | | using System.Linq.Expressions; |
| | | 4 | | using System.Runtime.ExceptionServices; |
| | | 5 | | |
| | | 6 | | namespace AsyncResponse; |
| | | 7 | | |
| | | 8 | | /// <summary> |
| | | 9 | | /// Runtime <see cref="IDurableFlowContext"/> bound to one execution of one flow run. Owns the |
| | | 10 | | /// checkpointed-flow mechanics so flow code doesn't have to: step guards, result memoization, the |
| | | 11 | | /// pending-correlation-id breadcrumb, fresh-start vs re-attach, and the durable resume/failure |
| | | 12 | | /// callbacks that point back at the flow executor. |
| | | 13 | | /// <para> |
| | | 14 | | /// Not thread-safe by design: a flow body runs sequentially, and <c>until</c> predicates run on |
| | | 15 | | /// the channel's dispatch path only while the flow itself is parked awaiting that same step. |
| | | 16 | | /// </para> |
| | | 17 | | /// </summary> |
| | | 18 | | internal sealed class DurableFlowContext : IDurableFlowContext |
| | | 19 | | { |
| | | 20 | | private readonly FlowState _state; |
| | | 21 | | private readonly IFlowStateStore _store; |
| | | 22 | | private readonly IAsyncResponseBuilder _builder; |
| | | 23 | | private readonly AsyncResponseContextPropagation _propagation; |
| | | 24 | | private readonly DurableFlowOptions _options; |
| | | 25 | | private readonly IAsyncResponseSubscriber _subscriber; |
| | | 26 | | private readonly IRecoverableAsyncResponseSubscriber? _recoverableSubscriber; |
| | | 27 | | private readonly ILogger _logger; |
| | | 28 | | private readonly FlowExecutionLease _lease; |
| | | 29 | | private readonly TimeProvider _timeProvider; |
| | | 30 | | private readonly IDurableFlowExecutionObserver[] _observers; |
| | | 31 | | private readonly IWorkerTransport? _workerTransport; |
| | | 32 | | private readonly TimeSpan? _channelDefaultWaitTimeout; |
| | | 33 | | private readonly CancellationToken _hostStopping; |
| | | 34 | | private bool _suspended; |
| | | 35 | | |
| | | 36 | | // The failure of a park that did not commit (see ParkAsync). Sticky like _suspended: flow |
| | | 37 | | // code that swallowed the first throw gets it again from every later context call, and from |
| | | 38 | | // the executor's flush when the body returns normally. |
| | | 39 | | private ExceptionDispatchInfo? _parkFailure; |
| | | 40 | | |
| | | 41 | | // Step names that RETURNED in this execution (memoized or freshly completed); see GetStep. |
| | | 42 | | private HashSet<string>? _returnedSteps; |
| | | 43 | | |
| | | 44 | | // 1 while a step call of this context is in flight; see EnterStep. |
| | | 45 | | private int _activeStep; |
| | | 46 | | |
| | | 47 | | // Identifies the context whose step is executing on the current async flow. Tells a step |
| | | 48 | | // called from INSIDE another step's body (nested: sequential, supported) from a sibling |
| | | 49 | | // started next to it (Task.WhenAll: concurrent, not supported) — the sibling starts from the |
| | | 50 | | // flow body's execution context, where this is not set. A bare token rather than the context: |
| | | 51 | | // execution-context snapshots outlive the execution (the in-memory transport keeps one with |
| | | 52 | | // every delayed wake-up), and must not pin the ledger with them. |
| | 13 | 53 | | private static readonly AsyncLocal<object?> ActiveStepOwner = new(); |
| | 2053 | 54 | | private readonly object _stepToken = new(); |
| | | 55 | | private bool _progressDirty; |
| | | 56 | | private DateTime _lastPersistenceUtc; |
| | | 57 | | |
| | | 58 | | // The next ledger-size estimate (in chars) that logs the growth warning; long.MaxValue when |
| | | 59 | | // the warning is disabled. Doubles after every warning so a long run logs O(log n) times. |
| | | 60 | | private long _nextLedgerSizeWarningChars; |
| | | 61 | | |
| | | 62 | | /// <summary> |
| | | 63 | | /// The deepest child-flow nesting a long park supports (see |
| | | 64 | | /// <see cref="ExtendAncestorLedgersAsync"/>): every ancestor up to the root is refreshed, and a |
| | | 65 | | /// chain longer than this fails the run terminally instead of being silently truncated — the |
| | | 66 | | /// previous 16-level cap stopped walking with the root unrefreshed, so a leaf nested 17 deep |
| | | 67 | | /// parked "successfully" while its root expired underneath it. Cycles are detected separately |
| | | 68 | | /// (a visited set), so this bounds only the cost of a legitimately absurd nesting. |
| | | 69 | | /// </summary> |
| | | 70 | | internal const int MaxAncestorLedgerDepth = 256; |
| | | 71 | | |
| | | 72 | | /// <summary>Creates the context for one execution of the given run.</summary> |
| | 2053 | 73 | | public DurableFlowContext( |
| | 2053 | 74 | | FlowState state, |
| | 2053 | 75 | | IFlowStateStore store, |
| | 2053 | 76 | | IAsyncResponseBuilder builder, |
| | 2053 | 77 | | AsyncResponseContextPropagation propagation, |
| | 2053 | 78 | | DurableFlowOptions options, |
| | 2053 | 79 | | IAsyncResponseSubscriber subscriber, |
| | 2053 | 80 | | IRecoverableAsyncResponseSubscriber? recoverableSubscriber, |
| | 2053 | 81 | | ILogger logger, |
| | 2053 | 82 | | FlowExecutionLease lease, |
| | 2053 | 83 | | TimeProvider? timeProvider = null, |
| | 2053 | 84 | | IDurableFlowExecutionObserver[]? observers = null, |
| | 2053 | 85 | | IWorkerTransport? workerTransport = null, |
| | 2053 | 86 | | TimeSpan? channelDefaultWaitTimeout = null, |
| | 2053 | 87 | | CancellationToken hostStopping = default) |
| | | 88 | | { |
| | 2053 | 89 | | _state = state; |
| | 2053 | 90 | | _store = store; |
| | 2053 | 91 | | _builder = builder; |
| | 2053 | 92 | | _propagation = propagation; |
| | 2053 | 93 | | _options = options; |
| | 2053 | 94 | | _nextLedgerSizeWarningChars = options.LedgerSizeWarningBytes ?? long.MaxValue; |
| | 2053 | 95 | | _subscriber = subscriber; |
| | 2053 | 96 | | _recoverableSubscriber = recoverableSubscriber; |
| | 2053 | 97 | | _logger = logger; |
| | 2053 | 98 | | _lease = lease; |
| | 2053 | 99 | | _timeProvider = timeProvider ?? TimeProvider.System; |
| | 2053 | 100 | | _observers = observers ?? []; |
| | 2053 | 101 | | _workerTransport = workerTransport; |
| | 2053 | 102 | | _channelDefaultWaitTimeout = channelDefaultWaitTimeout; |
| | 2053 | 103 | | _hostStopping = hostStopping; |
| | 2053 | 104 | | } |
| | | 105 | | |
| | 26851 | 106 | | private DateTime UtcNow => _timeProvider.GetUtcNow().UtcDateTime; |
| | | 107 | | |
| | | 108 | | /// <summary> |
| | | 109 | | /// Invokes every registered execution observer. Observers run on the execution path by |
| | | 110 | | /// contract: an observer exception fails this execution attempt exactly like a step failure |
| | | 111 | | /// (AsyncResponse.Testing injects deterministic crashes through precisely this). |
| | | 112 | | /// </summary> |
| | | 113 | | private async ValueTask NotifyStepAsync( |
| | | 114 | | Func<IDurableFlowExecutionObserver, DurableFlowStepEvent, ValueTask> invoke, |
| | | 115 | | string stepName, |
| | | 116 | | DurableFlowStepKind kind, |
| | | 117 | | string? correlationId = null, |
| | | 118 | | DateTime? wakeAtUtc = null) |
| | | 119 | | { |
| | 12602 | 120 | | if (_observers.Length == 0) |
| | 8680 | 121 | | return; |
| | | 122 | | |
| | 3922 | 123 | | var stepEvent = new DurableFlowStepEvent(FlowId, stepName, kind, correlationId, wakeAtUtc); |
| | 23430 | 124 | | foreach (var observer in _observers) |
| | 7808 | 125 | | await invoke(observer, stepEvent).ConfigureAwait(false); |
| | 12572 | 126 | | } |
| | | 127 | | |
| | 962 | 128 | | internal bool IsSuspended => _suspended; |
| | | 129 | | |
| | | 130 | | /// <inheritdoc /> |
| | 8762 | 131 | | public string FlowId => _state.FlowId!; |
| | | 132 | | |
| | | 133 | | /// <inheritdoc /> |
| | | 134 | | public async Task StepAsync(string name, Func<Task> step, CancellationToken cancellationToken = default) |
| | | 135 | | { |
| | 1908 | 136 | | ThrowIfSuspended(); |
| | 1906 | 137 | | ArgumentException.ThrowIfNullOrWhiteSpace(name); |
| | 1906 | 138 | | ArgumentNullException.ThrowIfNull(step); |
| | | 139 | | |
| | 1906 | 140 | | using var active = EnterStep(name); |
| | 1906 | 141 | | var checkpoint = GetStep(name); |
| | 1902 | 142 | | if (checkpoint.Completed) |
| | 30 | 143 | | return; |
| | | 144 | | |
| | 1872 | 145 | | cancellationToken.ThrowIfCancellationRequested(); |
| | 4100 | 146 | | await NotifyStepAsync(static (o, e) => o.OnStepStartingAsync(e), name, DurableFlowStepKind.Local).ConfigureAwait |
| | 1868 | 147 | | await step().ConfigureAwait(false); |
| | 1866 | 148 | | _lease.ThrowIfLost(); |
| | | 149 | | // Deliberately NOT the caller's token (awaited-step parity): the step's side effect has |
| | | 150 | | // already happened, so the checkpoint is its only record. A cancellation here lost the |
| | | 151 | | // checkpoint — the redelivered execution re-ran the side effect — and the store's |
| | | 152 | | // OperationCanceledException tripped MarkLost on a lease whose row was intact. |
| | 1866 | 153 | | await CompleteStepAsync(name, checkpoint, resultJson: null, CancellationToken.None).ConfigureAwait(false); |
| | 1890 | 154 | | } |
| | | 155 | | |
| | | 156 | | /// <inheritdoc /> |
| | | 157 | | public async Task<TResult> StepAsync<TResult>(string name, Func<Task<TResult>> step, CancellationToken cancellationT |
| | | 158 | | { |
| | 1839 | 159 | | ThrowIfSuspended(); |
| | 1839 | 160 | | ArgumentException.ThrowIfNullOrWhiteSpace(name); |
| | 1839 | 161 | | ArgumentNullException.ThrowIfNull(step); |
| | | 162 | | |
| | 1839 | 163 | | using var active = EnterStep(name); |
| | 1839 | 164 | | var checkpoint = GetStep(name); |
| | 1839 | 165 | | if (checkpoint.Completed) |
| | 54 | 166 | | return DeserializeResult<TResult>(checkpoint.ResultJson); |
| | | 167 | | |
| | 1785 | 168 | | cancellationToken.ThrowIfCancellationRequested(); |
| | 2613 | 169 | | await NotifyStepAsync(static (o, e) => o.OnStepStartingAsync(e), name, DurableFlowStepKind.Local).ConfigureAwait |
| | 1783 | 170 | | var result = await step().ConfigureAwait(false); |
| | 1783 | 171 | | _lease.ThrowIfLost(); |
| | | 172 | | // Not the caller's token: see the untyped overload above. |
| | 1783 | 173 | | await CompleteStepAsync(name, checkpoint, AsyncResponseJson.Serialize(result), CancellationToken.None).Configure |
| | 1781 | 174 | | return result; |
| | 1835 | 175 | | } |
| | | 176 | | |
| | | 177 | | /// <inheritdoc /> |
| | | 178 | | public Task DelayAsync(string name, TimeSpan delay, CancellationToken cancellationToken = default) |
| | | 179 | | { |
| | 184 | 180 | | ThrowIfSuspended(); |
| | 184 | 181 | | ArgumentException.ThrowIfNullOrWhiteSpace(name); |
| | | 182 | | |
| | 184 | 183 | | var checkpoint = GetStep(name); |
| | 184 | 184 | | if (checkpoint.Completed) |
| | 12 | 185 | | return Task.CompletedTask; |
| | | 186 | | |
| | | 187 | | // The due time anchors at the FIRST execution that reaches this step and is checkpointed; |
| | | 188 | | // replays (crash, redeploy, chunked wake-up) wait out the remainder, never restart. The |
| | | 189 | | // checkpointed instant therefore wins outright — the argument is not even looked at on a |
| | | 190 | | // replay, exactly as in DelayUntilAsync: an edit to the delay while a run is mid-sleep |
| | | 191 | | // must not change that run, and least of all fail it (validating the new argument here |
| | | 192 | | // turned a parked, perfectly valid timer into a terminal failure on resume). |
| | 172 | 193 | | if (checkpoint.WakeAtUtc is { } persisted) |
| | 74 | 194 | | return DelayCoreAsync(name, checkpoint, persisted, cancellationToken); |
| | | 195 | | |
| | | 196 | | // Fresh arrival: validate the requested span BEFORE the UtcNow.Add below, so |
| | | 197 | | // TimeSpan.MaxValue (or any absurd span) surfaces as the terminal sleep-ceiling failure |
| | | 198 | | // rather than the DateTime overflow the addition would throw first. |
| | 98 | 199 | | var requested = delay > TimeSpan.Zero ? delay : TimeSpan.Zero; |
| | 98 | 200 | | ThrowIfSleepBeyondLedger(name, requested); |
| | 94 | 201 | | return DelayCoreAsync(name, checkpoint, UtcNow.Add(requested), cancellationToken); |
| | | 202 | | } |
| | | 203 | | |
| | | 204 | | /// <inheritdoc /> |
| | | 205 | | public Task DelayUntilAsync(string name, DateTimeOffset wakeAtUtc, CancellationToken cancellationToken = default) |
| | | 206 | | { |
| | 2 | 207 | | ThrowIfSuspended(); |
| | 2 | 208 | | ArgumentException.ThrowIfNullOrWhiteSpace(name); |
| | | 209 | | |
| | 2 | 210 | | var checkpoint = GetStep(name); |
| | 2 | 211 | | if (checkpoint.Completed) |
| | 0 | 212 | | return Task.CompletedTask; |
| | | 213 | | |
| | | 214 | | // The checkpointed instant wins over the argument on replay, so a code edit that changes |
| | | 215 | | // the target while a run is mid-sleep cannot double- or under-sleep that run. |
| | 2 | 216 | | return DelayCoreAsync(name, checkpoint, checkpoint.WakeAtUtc ?? wakeAtUtc.UtcDateTime, cancellationToken); |
| | | 217 | | } |
| | | 218 | | |
| | | 219 | | private async Task DelayCoreAsync(string name, FlowStepState checkpoint, DateTime wakeAtUtc, CancellationToken cance |
| | | 220 | | { |
| | 170 | 221 | | using var active = EnterStep(name); |
| | 170 | 222 | | cancellationToken.ThrowIfCancellationRequested(); |
| | 406 | 223 | | await NotifyStepAsync(static (o, e) => o.OnStepStartingAsync(e), name, DurableFlowStepKind.Timer, wakeAtUtc: wak |
| | | 224 | | |
| | 166 | 225 | | var remaining = wakeAtUtc - UtcNow; |
| | 166 | 226 | | var firstPass = checkpoint.WakeAtUtc is null; |
| | | 227 | | |
| | | 228 | | // The ledger bound is settled at the FIRST arm, against the options in force then; a |
| | | 229 | | // replay never re-litigates the checkpointed sleep against the CURRENT options — raising |
| | | 230 | | // StateExpiry mid-sleep shrinks the recomputed bound and would terminally fail a parked |
| | | 231 | | // timer that was valid when it was armed, exactly the class of failure the |
| | | 232 | | // argument-ignoring contract above rules out. |
| | 166 | 233 | | if (firstPass) |
| | 92 | 234 | | ThrowIfSleepBeyondLedger(name, remaining); |
| | | 235 | | |
| | | 236 | | // The skew proof rides the wake-up that carried it, and that wake-up targets the ONE step |
| | | 237 | | // whose due time is already persisted — the parked timer this delivery re-executes. |
| | | 238 | | // Claiming it here (one-shot) scopes the forced-early fallback to that step alone: an |
| | | 239 | | // unrelated later timer in the same replay suspends normally instead of inheriting an |
| | | 240 | | // exemption that would pin it in process for its full remainder, or fail it on the |
| | | 241 | | // 49.7-day ceiling below. |
| | 166 | 242 | | var forcedEarly = !firstPass && WorkerJobSkewScope.TryConsumeForcedEarlyExecution(); |
| | | 243 | | |
| | 166 | 244 | | if (firstPass) |
| | | 245 | | { |
| | | 246 | | // Persist the breadcrumb BEFORE any wake-up can exist, with a TTL that covers the whole |
| | | 247 | | // sleep plus the normal idle margin — a run must never out-sleep its own ledger. |
| | 92 | 248 | | checkpoint.WakeAtUtc = wakeAtUtc; |
| | 92 | 249 | | checkpoint.Faulted = false; |
| | 92 | 250 | | checkpoint.Message = remaining > TimeSpan.Zero ? $"Sleeping until {wakeAtUtc:O}." : null; |
| | 92 | 251 | | await SaveForSleepAsync(remaining, cancellationToken).ConfigureAwait(false); |
| | | 252 | | } |
| | | 253 | | |
| | 158 | 254 | | if (remaining > TimeSpan.Zero) |
| | | 255 | | { |
| | 248 | 256 | | await NotifyStepAsync(static (o, e) => o.OnStepWaitingAsync(e), name, DurableFlowStepKind.Timer, wakeAtUtc: |
| | | 257 | | |
| | | 258 | | // MaxPublishDelay <= zero marks a transport whose delayed capability is unavailable in |
| | | 259 | | // the current configuration (an SQS FIFO worker queue): suspend-then-throw would strand |
| | | 260 | | // the run as "sleeping" with no wake-up, so treat it as not delayed-capable at all. |
| | | 261 | | // |
| | | 262 | | // The skew marker rules out suspension for a different reason: this delivery only |
| | | 263 | | // happened because the executor proved (over consecutive hops) that the transport's |
| | | 264 | | // delay gate and the stamping clock disagree. Suspending again would enqueue a FRESH |
| | | 265 | | // wake-up whose stall counters start at zero, discarding that proof and looping |
| | | 266 | | // forever — so the remainder is waited out in process instead, which honors the due |
| | | 267 | | // time. That fallback needs no new envelope and is the same one non-delayed transports |
| | | 268 | | // always take. |
| | 112 | 269 | | if (remaining > _options.TimerInProcessThreshold |
| | 112 | 270 | | && !forcedEarly |
| | 112 | 271 | | && _workerTransport is IDelayedWorkerTransport delayedTransport |
| | 112 | 272 | | && delayedTransport.MaxPublishDelay > TimeSpan.Zero) |
| | | 273 | | { |
| | | 274 | | // Suspend instead of waiting here: the delayed wake-up job re-executes the flow at |
| | | 275 | | // (or chunked toward) the due time, and this run holds no worker, lease, or memory |
| | | 276 | | // while it sleeps. Mirrors the child-flow suspension ordering: persist, enqueue, |
| | | 277 | | // throw — a crash between the persist and the enqueue leaves this job unacked, so |
| | | 278 | | // broker redelivery re-runs the step and re-enqueues the wake-up. |
| | 72 | 279 | | await SuspendForTimerAsync(name, wakeAtUtc, remaining, cancellationToken).ConfigureAwait(false); |
| | 0 | 280 | | throw new InvalidOperationException("Unreachable."); |
| | | 281 | | } |
| | | 282 | | |
| | | 283 | | // One delivery is never held past the in-process budget: a longer remainder is waited |
| | | 284 | | // in hops, each under a fresh delivery (see HandOverTimerAsync). |
| | 40 | 285 | | var wait = InProcessParkBudget() is { } budget && budget < remaining ? budget : remaining; |
| | | 286 | | |
| | 40 | 287 | | if (wait > AsyncResponseChannelOptions.MaxTimerBackedTimeout) |
| | | 288 | | { |
| | 0 | 289 | | throw new DurableFlowFailedException( |
| | 0 | 290 | | $"Timer step '{name}' of flow '{FlowId}' sleeps for {remaining.TotalDays:0.#} days, which exceeds th |
| | 0 | 291 | | $"{AsyncResponseChannelOptions.MaxTimerBackedTimeout.TotalDays:0.#}-day .NET timer ceiling, and " + |
| | 0 | 292 | | (forcedEarly |
| | 0 | 293 | | ? "this wake-up was released early because the transport's delay gate and the publishing clock d |
| | 0 | 294 | | "re-suspending would loop instead of sleeping. Fix the clock skew between the application and |
| | 0 | 295 | | : $"the registered worker transport has no native delayed delivery ({nameof(IDelayedWorkerTransp |
| | 0 | 296 | | "Use a delayed-capable transport (in-memory, Azure Service Bus, SQS, PostgreSQL, SQL Server, M |
| | | 297 | | } |
| | | 298 | | |
| | 40 | 299 | | if (!firstPass) |
| | | 300 | | { |
| | | 301 | | // Replayed execution about to resume the sleep in process: the executor's |
| | | 302 | | // unconditional per-attempt save reset the ledger TTL to StateExpiry, and every |
| | | 303 | | // store recomputes expiry from "now" — a resumed sleep longer than StateExpiry |
| | | 304 | | // would out-sleep its own ledger and be silently dropped mid-wait. Re-extend to |
| | | 305 | | // cover the remainder (the suspend path re-extends every pass in SuspendForTimerAsync). |
| | 16 | 306 | | await SaveForSleepAsync(remaining, cancellationToken).ConfigureAwait(false); |
| | | 307 | | } |
| | | 308 | | |
| | 40 | 309 | | await WaitInProcessAsync(wait, cancellationToken).ConfigureAwait(false); |
| | 38 | 310 | | _lease.ThrowIfLost(); |
| | | 311 | | |
| | 38 | 312 | | if (wait < remaining) |
| | | 313 | | { |
| | | 314 | | // Measured again rather than computed: a timer that fired late may already have |
| | | 315 | | // covered the rest, and a due timer completes here like any other. |
| | 22 | 316 | | var left = wakeAtUtc - UtcNow; |
| | 22 | 317 | | if (left > TimeSpan.Zero) |
| | | 318 | | { |
| | 22 | 319 | | await HandOverTimerAsync(name, wakeAtUtc, left, cancellationToken).ConfigureAwait(false); |
| | 0 | 320 | | throw new InvalidOperationException("Unreachable."); |
| | | 321 | | } |
| | | 322 | | } |
| | | 323 | | } |
| | | 324 | | |
| | 62 | 325 | | await CompleteStepAsync(name, checkpoint, resultJson: null, CancellationToken.None, kind: DurableFlowStepKind.Ti |
| | 58 | 326 | | } |
| | | 327 | | |
| | | 328 | | /// <summary> |
| | | 329 | | /// The longest an in-process timer wait may hold one delivery, or <c>null</c> for no bound: |
| | | 330 | | /// half of the in-flight ceiling the worker transport advertises |
| | | 331 | | /// (<see cref="IWorkerTransportInFlightLimit"/>) — the other half is headroom for the steps |
| | | 332 | | /// that ran before the timer in the same delivery and for the hand-over itself — shortened |
| | | 333 | | /// further by <see cref="DurableFlowOptions.MaxInProcessParkDuration"/>, which also supplies a |
| | | 334 | | /// bound when the transport advertises none. Capped at the BCL timer ceiling the wait arms. |
| | | 335 | | /// </summary> |
| | | 336 | | private TimeSpan? InProcessParkBudget() |
| | | 337 | | { |
| | 407 | 338 | | TimeSpan? budget = _workerTransport is IWorkerTransportInFlightLimit { MaxInFlightDuration: { } ceiling } && cei |
| | 407 | 339 | | ? ceiling / 2 |
| | 407 | 340 | | : null; |
| | | 341 | | |
| | 407 | 342 | | if (_options.MaxInProcessParkDuration is { } configured && (budget is null || configured < budget)) |
| | 6 | 343 | | budget = configured; |
| | | 344 | | |
| | 407 | 345 | | return budget > AsyncResponseChannelOptions.MaxTimerBackedTimeout |
| | 407 | 346 | | ? AsyncResponseChannelOptions.MaxTimerBackedTimeout |
| | 407 | 347 | | : budget; |
| | | 348 | | } |
| | | 349 | | |
| | | 350 | | /// <summary> |
| | | 351 | | /// In-process timer wait under the execution lease — the fallback for transports without |
| | | 352 | | /// delayed delivery and for sub-threshold remainders. Cancellation (caller token, lease loss, |
| | | 353 | | /// host stop) deliberately leaves the checkpoint untouched: the persisted due time is the |
| | | 354 | | /// breadcrumb, and the redelivered execution waits out the remainder — the timer itself |
| | | 355 | | /// cannot fault. |
| | | 356 | | /// <para> |
| | | 357 | | /// Host stop is wired in explicitly. Nothing else ends this wait at shutdown — the lease keeps |
| | | 358 | | /// renewing for as long as the process lives — so a deploy used to wait a parked handler out |
| | | 359 | | /// for the transport's whole drain budget and then kill it. |
| | | 360 | | /// </para> |
| | | 361 | | /// </summary> |
| | | 362 | | private async Task WaitInProcessAsync(TimeSpan wait, CancellationToken cancellationToken) |
| | | 363 | | { |
| | 40 | 364 | | using var linked = cancellationToken.CanBeCanceled || _hostStopping.CanBeCanceled |
| | 40 | 365 | | ? CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, _lease.LostToken, _hostStopping) |
| | 40 | 366 | | : null; |
| | | 367 | | |
| | | 368 | | try |
| | | 369 | | { |
| | 40 | 370 | | await Task.Delay(wait, _timeProvider, linked?.Token ?? _lease.LostToken).ConfigureAwait(false); |
| | 38 | 371 | | } |
| | 2 | 372 | | catch (OperationCanceledException ex) |
| | | 373 | | { |
| | 2 | 374 | | _lease.ThrowIfLost(ex); |
| | 2 | 375 | | if (_hostStopping.IsCancellationRequested && !cancellationToken.IsCancellationRequested) |
| | 2 | 376 | | throw HostStopping(ex); |
| | 0 | 377 | | throw; |
| | | 378 | | } |
| | 38 | 379 | | } |
| | | 380 | | |
| | | 381 | | /// <summary> |
| | | 382 | | /// The exception an in-process park ends with at host stop. A cancellation on purpose (the |
| | | 383 | | /// executor's lease-contention poll does the same): the worker transport treats the job as not |
| | | 384 | | /// executed and redelivers it after the restart, and neither the step nor the run is faulted. |
| | | 385 | | /// </summary> |
| | | 386 | | private DurableFlowInterruptedException HostStopping(Exception cause) |
| | 4 | 387 | | => new($"Host is stopping; durable flow '{FlowId}' left its in-process wait and the delivery is abandoned for re |
| | | 388 | | |
| | | 389 | | private Task SuspendForTimerAsync(string name, DateTime wakeAtUtc, TimeSpan remaining, CancellationToken cancellatio |
| | | 390 | | { |
| | 72 | 391 | | _state.LastMessage = $"Flow {FlowId} sleeping until {wakeAtUtc:O} at step '{name}'."; |
| | 72 | 392 | | var id = FlowId; |
| | 72 | 393 | | return ParkAsync( |
| | 72 | 394 | | remaining, |
| | 72 | 395 | | () => _builder.EnqueueWorkerAsync<IDurableFlowExecutor>(executor => executor.ExecuteAsync(id), remaining), |
| | 72 | 396 | | cancellationToken); |
| | | 397 | | } |
| | | 398 | | |
| | | 399 | | /// <summary> |
| | | 400 | | /// Ends an in-process hop with time still to sleep: checkpoint, publish an IMMEDIATE wake-up, |
| | | 401 | | /// suspend. The wake-up replays to this timer — its due time is checkpointed — and waits the |
| | | 402 | | /// next hop under a new delivery, whose in-flight clock the broker starts from zero. Only ever |
| | | 403 | | /// called AFTER a hop was waited: the wake-up is not delayed, so publishing it without having |
| | | 404 | | /// waited would spin deliveries instead of sleeping. |
| | | 405 | | /// </summary> |
| | | 406 | | private Task HandOverTimerAsync(string name, DateTime wakeAtUtc, TimeSpan left, CancellationToken cancellationToken) |
| | | 407 | | { |
| | 22 | 408 | | _state.LastMessage = $"Flow {FlowId} sleeping until {wakeAtUtc:O} at step '{name}' (continuing under a fresh del |
| | 22 | 409 | | var id = FlowId; |
| | 22 | 410 | | return ParkAsync( |
| | 22 | 411 | | left, |
| | 22 | 412 | | () => _builder.EnqueueWorkerAsync<IDurableFlowExecutor>(executor => executor.ExecuteAsync(id)), |
| | 22 | 413 | | cancellationToken); |
| | | 414 | | } |
| | | 415 | | |
| | | 416 | | /// <summary> |
| | | 417 | | /// Commits a park — persist, publish the wake-up — and only then marks this execution |
| | | 418 | | /// suspended. The executor acknowledges a suspended execution's delivery, so the flag must |
| | | 419 | | /// never be up before the wake-up exists: raised first (as it used to be), a failed save or |
| | | 420 | | /// publish left it set, flow code that catches <see cref="Exception"/> around the step carried |
| | | 421 | | /// on into the next context call, that call threw "suspended", and the delivery was |
| | | 422 | | /// acknowledged for a run nothing would ever wake — on the child path with a child ledger that |
| | | 423 | | /// was never enqueued. A failure is kept and surfaced again (see <see cref="ThrowIfSuspended"/> |
| | | 424 | | /// and <see cref="FlushProgressAsync"/>) so the attempt ends as the retriable failure it is |
| | | 425 | | /// even when the first throw was swallowed. |
| | | 426 | | /// </summary> |
| | | 427 | | private async Task ParkAsync(TimeSpan window, Func<Task> publishWakeUp, CancellationToken cancellationToken) |
| | | 428 | | { |
| | | 429 | | try |
| | | 430 | | { |
| | 212 | 431 | | await SaveForSleepAsync(window, cancellationToken).ConfigureAwait(false); |
| | 212 | 432 | | await publishWakeUp().ConfigureAwait(false); |
| | 208 | 433 | | } |
| | 4 | 434 | | catch (Exception ex) |
| | | 435 | | { |
| | 4 | 436 | | _parkFailure = ExceptionDispatchInfo.Capture(ex); |
| | 4 | 437 | | throw; |
| | | 438 | | } |
| | | 439 | | |
| | 208 | 440 | | _suspended = true; |
| | 208 | 441 | | throw new DurableFlowSuspendedException(_state.LastMessage ?? $"Flow {FlowId} is suspended."); |
| | | 442 | | } |
| | | 443 | | |
| | | 444 | | /// <summary> |
| | | 445 | | /// The longest sleep a run's ledger can survive: the persistence ceiling minus the configured |
| | | 446 | | /// <see cref="DurableFlowOptions.StateExpiry"/>, so the TTL stamped by |
| | | 447 | | /// <see cref="SaveForSleepAsync"/> (<c>sleep + StateExpiry</c>) always fits the ceiling with |
| | | 448 | | /// the full idle margin intact. Allowing sleeps up to the ceiling itself would stamp a TTL |
| | | 449 | | /// that expires exactly at the due instant — any wake latency or store clock skew then finds |
| | | 450 | | /// the flow state already gone and strands the run unfinished. |
| | | 451 | | /// </summary> |
| | | 452 | | private void ThrowIfSleepBeyondLedger(string name, TimeSpan sleep) |
| | | 453 | | { |
| | 190 | 454 | | var maxSleep = AsyncResponseChannelOptions.MaxPersistenceTtl - _options.StateExpiry; |
| | 190 | 455 | | if (sleep <= maxSleep) |
| | 186 | 456 | | return; |
| | | 457 | | |
| | 4 | 458 | | throw new DurableFlowFailedException( |
| | 4 | 459 | | $"Timer step '{name}' of flow '{FlowId}' sleeps for {sleep.TotalDays:0} days; the maximum is " + |
| | 4 | 460 | | $"{maxSleep.TotalDays:0} days — the {AsyncResponseChannelOptions.MaxPersistenceTtl.TotalDays:0}-day persiste |
| | 4 | 461 | | $"{nameof(DurableFlowOptions)}.{nameof(DurableFlowOptions.StateExpiry)} ({_options.StateExpiry.TotalDays:0.# |
| | 4 | 462 | | "stamps are computed as \"now + sleep + StateExpiry\" and the ledger must outlive its own wake-up."); |
| | | 463 | | } |
| | | 464 | | |
| | | 465 | | /// <summary> |
| | | 466 | | /// Checkpoint save whose TTL covers a known wait window (a timer's sleep, or an awaited step's |
| | | 467 | | /// timeout): <c>remaining + StateExpiry</c>, saturated at the persistence ceiling. The ordinary |
| | | 468 | | /// <see cref="SaveAsync"/> TTL bounds <em>idle</em> time between checkpoints; a run parked on a |
| | | 469 | | /// timer or an awaited response is idle by design for the whole window — and because |
| | | 470 | | /// <see cref="ThrowIfSleepBeyondLedger"/> caps every sleep at ceiling − StateExpiry (and step |
| | | 471 | | /// timeouts are timer-bounded far below it), the sum here always carries the full StateExpiry |
| | | 472 | | /// margin past the due instant. |
| | | 473 | | /// </summary> |
| | | 474 | | private async Task SaveForSleepAsync(TimeSpan remaining, CancellationToken cancellationToken) |
| | | 475 | | { |
| | 1815 | 476 | | var margin = AsyncResponseChannelOptions.MaxPersistenceTtl - _options.StateExpiry; |
| | 1815 | 477 | | var ttl = remaining <= TimeSpan.Zero |
| | 1815 | 478 | | ? _options.StateExpiry |
| | 1815 | 479 | | : remaining >= margin |
| | 1815 | 480 | | ? AsyncResponseChannelOptions.MaxPersistenceTtl |
| | 1815 | 481 | | : remaining + _options.StateExpiry; |
| | | 482 | | |
| | | 483 | | // The wait outlives this save's own TTL stamp only if every later write of this ledger |
| | | 484 | | // carries it forward — a spurious early redelivery of the parked run stamps the plain |
| | | 485 | | // StateExpiry in the executor's per-attempt save before it replays back here. The floor |
| | | 486 | | // in the ledger is what those writes honor (FlowStateRetention.EffectiveTtl). |
| | 1815 | 487 | | if (ttl > _options.StateExpiry) |
| | 1697 | 488 | | FlowStateRetention.RaiseFloor(_state, UtcNow, ttl); |
| | 1815 | 489 | | await SaveAsync(cancellationToken, ttl: ttl).ConfigureAwait(false); |
| | | 490 | | |
| | | 491 | | // A parked ancestor's row must survive this run's whole wait, not just its own idle |
| | | 492 | | // margin: nothing refreshes an ancestor while it waits on this chain (lease renewal only |
| | | 493 | | // stamps the lease columns), so a descendant parking beyond the ancestor's StateExpiry |
| | | 494 | | // silently expired the ancestor and the eventual completion wake-up found no state. |
| | | 495 | | // Part of the park, not insurance around it: a failure here propagates BEFORE any wake-up |
| | | 496 | | // is published (every caller publishes after this save), so the delivery is retried from |
| | | 497 | | // the checkpoint above instead of the run parking on an ancestor that will expire under it. |
| | 1813 | 498 | | if (ttl > _options.StateExpiry && _state.ParentFlowId is not null) |
| | 34 | 499 | | await ExtendAncestorLedgersAsync(ttl, cancellationToken).ConfigureAwait(false); |
| | 1805 | 500 | | } |
| | | 501 | | |
| | | 502 | | /// <summary> |
| | | 503 | | /// Retention extension of the WHOLE ancestor chain when this run parks for a window its own |
| | | 504 | | /// plain <see cref="DurableFlowOptions.StateExpiry"/> would not cover. Each |
| | | 505 | | /// <see cref="FlowRunStatus.Running"/> ancestor gets its <see cref="FlowState.RetainUntilUtc"/> |
| | | 506 | | /// floor raised to cover the wait and its row re-stamped with the wait's TTL — a terminal or |
| | | 507 | | /// operator-suspended run is not waiting on this chain, and an absent row is never resurrected |
| | | 508 | | /// (the walk stops there and the expired-ancestor failure surfaces on wake-up, as before). A |
| | | 509 | | /// store failure PROPAGATES: the callers all publish their wake-up only after this returns, so |
| | | 510 | | /// the park fails with nothing published and the transport redelivers the execution, which |
| | | 511 | | /// replays to the same step and retries the chain. Swallowing it (an earlier behavior) let the |
| | | 512 | | /// child park "successfully" — wake-up and all — while the parent it would eventually complete |
| | | 513 | | /// into expired mid-wait, after which every step past the parent's child-await was lost with |
| | | 514 | | /// the parent's checkpoints. The chain is walked to the root with cycle detection; a chain |
| | | 515 | | /// that revisits an id or exceeds <see cref="MaxAncestorLedgerDepth"/> fails the run terminally |
| | | 516 | | /// (deterministic on every replay) rather than being truncated in silence. |
| | | 517 | | /// <para> |
| | | 518 | | /// A LOST compare-and-swap is not success. The previous design treated it as one — "a |
| | | 519 | | /// concurrent writer means the ancestor is alive and re-stamping its own expiry" — but the |
| | | 520 | | /// competing write was computed without this park in view: the parent replaying its |
| | | 521 | | /// child-await from a snapshot taken before this run persisted its sleep stamps the plain |
| | | 522 | | /// StateExpiry, and the executor's per-attempt save always does. Either one left the parent's |
| | | 523 | | /// row expiring under a wait this run had just parked into, with its wake-up published. So the |
| | | 524 | | /// ancestor is re-read after a lost race: when the write that won already carries a floor |
| | | 525 | | /// reaching this park (another extension of the same chain, or an earlier attempt of this |
| | | 526 | | /// one), the retention is proven and the walk moves on; otherwise the extension is retried |
| | | 527 | | /// against the new revision, a bounded number of times. Every write here still advances the |
| | | 528 | | /// ancestor's revision — it has to, the floor lives in the ledger — so a retry can cost an |
| | | 529 | | /// actively-executing ancestor one checkpoint (its next save loses the compare-and-swap and |
| | | 530 | | /// its delivery replays from the last one, now carrying the floor). That is the price of the |
| | | 531 | | /// guarantee; the earlier eight-attempt <see cref="FlowStateConcurrency.MutateAsync"/> fight |
| | | 532 | | /// was avoided by ceding the race, and ceding it is what lost the parent. The attempt bound |
| | | 533 | | /// keeps the fight finite: losing every attempt abandons the park (nothing published) so the |
| | | 534 | | /// delivery retries it later, exactly like a store failure. |
| | | 535 | | /// </para> |
| | | 536 | | /// <para> |
| | | 537 | | /// Every write of the ancestor after this one carries the floor forward (see |
| | | 538 | | /// <see cref="FlowStateRetention"/>), so the extension has to land once, not win every race |
| | | 539 | | /// from here to the wake-up. |
| | | 540 | | /// </para> |
| | | 541 | | /// </summary> |
| | | 542 | | private async Task ExtendAncestorLedgersAsync(TimeSpan ttl, CancellationToken cancellationToken) |
| | | 543 | | { |
| | 40 | 544 | | var visited = new HashSet<string>(StringComparer.Ordinal) { FlowId }; |
| | 40 | 545 | | var ancestorId = _state.ParentFlowId; |
| | 672 | 546 | | while (ancestorId is not null) |
| | | 547 | | { |
| | 642 | 548 | | if (!visited.Add(ancestorId)) |
| | | 549 | | { |
| | | 550 | | // Corrupted ledgers (ParentFlowId loops back into the chain). Deterministic on |
| | | 551 | | // every replay, so terminal: parking would leave the run waiting on ancestors |
| | | 552 | | // whose retention can never be established. |
| | 2 | 553 | | throw new DurableFlowFailedException( |
| | 2 | 554 | | $"Flow '{FlowId}' cannot park for {ttl}: its ancestor chain revisits flow '{ancestorId}' (a cycle in |
| | 2 | 555 | | "so the ledgers it would wait on cannot be kept alive. The stored ledgers are inconsistent; the run |
| | | 556 | | } |
| | | 557 | | |
| | 640 | 558 | | if (visited.Count > MaxAncestorLedgerDepth + 1) |
| | | 559 | | { |
| | 2 | 560 | | throw new DurableFlowFailedException( |
| | 2 | 561 | | $"Flow '{FlowId}' cannot park for {ttl}: it is nested more than {MaxAncestorLedgerDepth} child flows |
| | 2 | 562 | | "must be kept alive for the wait. Flatten the nesting."); |
| | | 563 | | } |
| | | 564 | | |
| | | 565 | | try |
| | | 566 | | { |
| | 638 | 567 | | ancestorId = await ExtendOneAncestorAsync(ancestorId, ttl, cancellationToken).ConfigureAwait(false); |
| | 632 | 568 | | } |
| | 6 | 569 | | catch (Exception ex) when (ex is not OperationCanceledException) |
| | | 570 | | { |
| | | 571 | | // Logged with the chain context, then rethrown AS IS: the park is abandoned with |
| | | 572 | | // nothing published, the delivery retries, and the store's own exception type |
| | | 573 | | // stays visible to whoever classifies it upstream. |
| | 6 | 574 | | _logger.LogWarning( |
| | 6 | 575 | | ex, |
| | 6 | 576 | | "Flow {FlowId} could not extend ancestor flow {AncestorFlowId}'s ledger retention for its {Ttl} park |
| | 6 | 577 | | FlowId, ancestorId, ttl); |
| | 6 | 578 | | throw; |
| | | 579 | | } |
| | | 580 | | } |
| | 30 | 581 | | } |
| | | 582 | | |
| | | 583 | | /// <summary> |
| | | 584 | | /// How many times one ancestor's extension is retried against a revision a concurrent write |
| | | 585 | | /// took. Each attempt re-reads the ancestor first, and a floor already reaching the park ends |
| | | 586 | | /// the attempt without a write. |
| | | 587 | | /// </summary> |
| | | 588 | | internal const int MaxAncestorExtensionAttempts = 4; |
| | | 589 | | |
| | | 590 | | /// <summary> |
| | | 591 | | /// Extends one ancestor (see <see cref="ExtendAncestorLedgersAsync"/>) and returns the id of |
| | | 592 | | /// the next ancestor up, or <c>null</c> when the walk stops here: the row is gone, the run is |
| | | 593 | | /// not <see cref="FlowRunStatus.Running"/>, or it has no parent. |
| | | 594 | | /// </summary> |
| | | 595 | | private async Task<string?> ExtendOneAncestorAsync(string ancestorId, TimeSpan ttl, CancellationToken cancellationTo |
| | | 596 | | { |
| | 654 | 597 | | for (var attempt = 1; ; attempt++) |
| | | 598 | | { |
| | 654 | 599 | | var ancestor = await _store.LoadAsync(ancestorId, cancellationToken).ConfigureAwait(false); |
| | 654 | 600 | | if (ancestor is null) |
| | | 601 | | { |
| | 0 | 602 | | _logger.LogWarning( |
| | 0 | 603 | | "Flow {FlowId} parked for {Ttl} but ancestor flow {AncestorFlowId} has no state (expired or deleted) |
| | 0 | 604 | | FlowId, ttl, ancestorId); |
| | 0 | 605 | | return null; |
| | | 606 | | } |
| | | 607 | | |
| | 654 | 608 | | if (ancestor.Status != FlowRunStatus.Running) |
| | 2 | 609 | | return null; |
| | | 610 | | |
| | 652 | 611 | | var now = UtcNow; |
| | 652 | 612 | | var until = FlowStateRetention.FloorAt(now, ttl); |
| | 652 | 613 | | if (FlowStateRetention.Covers(ancestor, until)) |
| | | 614 | | { |
| | | 615 | | // Proven by the re-read: whoever wrote last carried a floor reaching this park |
| | | 616 | | // (the write that beat a previous attempt, or a sibling park on the same chain). |
| | 56 | 617 | | return ancestor.ParentFlowId; |
| | | 618 | | } |
| | | 619 | | |
| | 596 | 620 | | FlowStateRetention.RaiseFloor(ancestor, now, ttl); |
| | 596 | 621 | | var expectedRevision = ancestor.Revision; |
| | 596 | 622 | | ancestor.Revision = checked(expectedRevision + 1); |
| | 596 | 623 | | ancestor.UpdatedAtUtc = now; |
| | 596 | 624 | | if (await _store.TryUpdateAsync( |
| | 596 | 625 | | ancestorId, |
| | 596 | 626 | | ancestor, |
| | 596 | 627 | | expectedRevision, |
| | 596 | 628 | | FlowStateRetention.EffectiveTtl(ancestor, ttl, now), |
| | 596 | 629 | | leaseId: null, |
| | 596 | 630 | | cancellationToken).ConfigureAwait(false)) |
| | 574 | 631 | | return ancestor.ParentFlowId; |
| | | 632 | | |
| | 20 | 633 | | if (attempt >= MaxAncestorExtensionAttempts) |
| | | 634 | | { |
| | | 635 | | // Not terminal: the ancestor is being written continuously right now, and the |
| | | 636 | | // next replay of this step may find it quiet. The park is abandoned with nothing |
| | | 637 | | // published, so the delivery retries it — the same route a store failure takes. |
| | 4 | 638 | | throw new InvalidOperationException( |
| | 4 | 639 | | $"Flow '{FlowId}' could not extend ancestor flow '{ancestorId}'s ledger retention for its {ttl} park |
| | 4 | 640 | | $"a concurrent write advanced the ancestor's revision on each of {attempt} attempts. The park is aba |
| | | 641 | | } |
| | | 642 | | |
| | 16 | 643 | | _logger.LogDebug( |
| | 16 | 644 | | "Flow {FlowId} lost the revision race extending ancestor flow {AncestorFlowId}'s ledger retention (attem |
| | 16 | 645 | | FlowId, ancestorId, attempt); |
| | 16 | 646 | | } |
| | 632 | 647 | | } |
| | | 648 | | |
| | | 649 | | /// <inheritdoc /> |
| | | 650 | | public Task<TResponse> AwaitStepAsync<TResponse>( |
| | | 651 | | string name, |
| | | 652 | | Func<string, Task> trigger, |
| | | 653 | | TimeSpan? timeout = null, |
| | | 654 | | CancellationToken cancellationToken = default) where TResponse : IAsyncResponsePayload |
| | 182 | 655 | | => AwaitStepCoreAsync<TResponse>(name, trigger, until: null, timeout, cancellationToken); |
| | | 656 | | |
| | | 657 | | /// <inheritdoc /> |
| | | 658 | | public Task<TResponse> AwaitStepAsync<TResponse>( |
| | | 659 | | string name, |
| | | 660 | | Func<string, Task> trigger, |
| | | 661 | | Func<TResponse, bool> until, |
| | | 662 | | TimeSpan? timeout = null, |
| | | 663 | | CancellationToken cancellationToken = default) where TResponse : IAsyncResponsePayload |
| | | 664 | | { |
| | 1411 | 665 | | ArgumentNullException.ThrowIfNull(until); |
| | 3459 | 666 | | return AwaitStepCoreAsync<TResponse>(name, trigger, payload => new ValueTask<bool>(until(payload)), timeout, can |
| | | 667 | | } |
| | | 668 | | |
| | | 669 | | /// <inheritdoc /> |
| | | 670 | | public Task<TResponse> AwaitStepAsync<TResponse>( |
| | | 671 | | string name, |
| | | 672 | | Func<string, Task> trigger, |
| | | 673 | | Func<TResponse, Task<bool>> until, |
| | | 674 | | TimeSpan? timeout = null, |
| | | 675 | | CancellationToken cancellationToken = default) where TResponse : IAsyncResponsePayload |
| | | 676 | | { |
| | 28 | 677 | | ArgumentNullException.ThrowIfNull(until); |
| | 66 | 678 | | return AwaitStepCoreAsync<TResponse>(name, trigger, payload => new ValueTask<bool>(until(payload)), timeout, can |
| | | 679 | | } |
| | | 680 | | |
| | | 681 | | /// <inheritdoc /> |
| | | 682 | | public Task ReportProgressAsync(string message, CancellationToken cancellationToken = default) |
| | | 683 | | { |
| | 30 | 684 | | ThrowIfSuspended(); |
| | 30 | 685 | | _state.LastMessage = message; |
| | 30 | 686 | | var now = UtcNow; |
| | 30 | 687 | | if (_options.ProgressPersistenceInterval <= TimeSpan.Zero |
| | 30 | 688 | | || now - _lastPersistenceUtc >= _options.ProgressPersistenceInterval) |
| | 2 | 689 | | return SaveAsync(cancellationToken); |
| | | 690 | | |
| | 28 | 691 | | _progressDirty = true; |
| | 28 | 692 | | return Task.CompletedTask; |
| | | 693 | | } |
| | | 694 | | |
| | | 695 | | /// <inheritdoc /> |
| | | 696 | | public TValue? GetValue<TValue>(string key) |
| | | 697 | | { |
| | 16 | 698 | | ThrowIfSuspended(); |
| | 16 | 699 | | ArgumentException.ThrowIfNullOrWhiteSpace(key); |
| | 16 | 700 | | return _state.Values is not null && _state.Values.TryGetValue(key, out var json) |
| | 16 | 701 | | ? JsonSafety.SafeDeserialize<TValue>(json) |
| | 16 | 702 | | : default; |
| | | 703 | | } |
| | | 704 | | |
| | | 705 | | /// <inheritdoc /> |
| | | 706 | | public Task SetValueAsync<TValue>(string key, TValue value, CancellationToken cancellationToken = default) |
| | | 707 | | { |
| | 1386 | 708 | | ThrowIfSuspended(); |
| | 1386 | 709 | | ArgumentException.ThrowIfNullOrWhiteSpace(key); |
| | 1386 | 710 | | var values = _state.Values ??= new Dictionary<string, string>(StringComparer.Ordinal); |
| | 1386 | 711 | | values[key] = AsyncResponseJson.Serialize(value); |
| | 1386 | 712 | | return SaveAsync(cancellationToken); |
| | | 713 | | } |
| | | 714 | | |
| | | 715 | | /// <inheritdoc /> |
| | | 716 | | public async Task<FlowState> AwaitChildFlowAsync<[DynamicallyAccessedMembers(DynamicallyAccessedMemberTypes.PublicCo |
| | | 717 | | string name, |
| | | 718 | | TInput input, |
| | | 719 | | string? flowId = null, |
| | | 720 | | bool failOnChildFailure = true, |
| | | 721 | | CancellationToken cancellationToken = default) |
| | | 722 | | where TFlow : class, IDurableFlow<TInput> |
| | | 723 | | { |
| | 296 | 724 | | ThrowIfSuspended(); |
| | 296 | 725 | | ArgumentException.ThrowIfNullOrWhiteSpace(name); |
| | 296 | 726 | | ArgumentNullException.ThrowIfNull(input); |
| | 296 | 727 | | if (flowId is not null) |
| | 12 | 728 | | ArgumentException.ThrowIfNullOrWhiteSpace(flowId); |
| | | 729 | | |
| | 296 | 730 | | using var active = EnterStep(name); |
| | 296 | 731 | | var checkpoint = GetStep(name); |
| | 296 | 732 | | var requestedChildFlowId = flowId ?? $"{FlowId}:{name}"; |
| | 296 | 733 | | var breadcrumb = checkpoint.ChildFlowId; |
| | 296 | 734 | | if (breadcrumb is not null && !string.Equals(breadcrumb, requestedChildFlowId, StringComparison.Ordinal)) |
| | | 735 | | { |
| | 2 | 736 | | throw new DurableFlowFailedException( |
| | 2 | 737 | | $"Step '{name}' of flow '{FlowId}' is already bound to child flow id '{breadcrumb}', " + |
| | 2 | 738 | | $"but this execution requested '{requestedChildFlowId}'. A durable step must keep the same child id on e |
| | | 739 | | } |
| | | 740 | | |
| | 294 | 741 | | var childFlowId = breadcrumb ?? requestedChildFlowId; |
| | 294 | 742 | | if (FlowStateConcurrency.FlowIdNotPortable(childFlowId) is { } rejection) |
| | | 743 | | { |
| | | 744 | | // Deterministic on every replay, so terminal rather than retriable: the composed id |
| | | 745 | | // can never become portable, and the constrained stores would reject the child row |
| | | 746 | | // anyway after a full budget of wasted redeliveries. |
| | 2 | 747 | | throw new DurableFlowFailedException( |
| | 2 | 748 | | $"Step '{name}' of flow '{FlowId}' composed a non-portable child flow id. {rejection}"); |
| | | 749 | | } |
| | | 750 | | |
| | 292 | 751 | | var inputJson = AsyncResponseJson.Serialize(input); |
| | 292 | 752 | | if (checkpoint.Completed) |
| | | 753 | | { |
| | 50 | 754 | | var completedChild = DeserializeResult<FlowState>(checkpoint.ResultJson) |
| | 50 | 755 | | ?? throw new DurableFlowFailedException( |
| | 50 | 756 | | $"Completed child step '{name}' of flow '{FlowId}' has no child-state snapshot."); |
| | 50 | 757 | | ThrowIfChildMismatched<TFlow, TInput>(completedChild, childFlowId, name, inputJson, completed: true); |
| | 50 | 758 | | ThrowIfChildFailed(completedChild, failOnChildFailure); |
| | 50 | 759 | | return completedChild; |
| | | 760 | | } |
| | | 761 | | |
| | 622 | 762 | | await NotifyStepAsync(static (o, e) => o.OnStepStartingAsync(e), name, DurableFlowStepKind.ChildFlow).ConfigureA |
| | | 763 | | |
| | 242 | 764 | | var child = await _store.LoadAsync(childFlowId, cancellationToken).ConfigureAwait(false); |
| | 242 | 765 | | if (child is null) |
| | | 766 | | { |
| | 114 | 767 | | if (breadcrumb is not null) |
| | | 768 | | { |
| | | 769 | | // The breadcrumb is persisted only after the child state exists, so a missing child |
| | | 770 | | // here means its ledger expired (StateExpiry) or was deleted while this parent was |
| | | 771 | | // suspended. Its outcome is unknowable; re-running it blind would re-execute side |
| | | 772 | | // effects of a possibly-completed run. Fail deterministically instead. |
| | 2 | 773 | | throw new DurableFlowFailedException( |
| | 2 | 774 | | $"Child flow '{childFlowId}' has no state (expired or deleted) while parent flow '{FlowId}' was wait |
| | 2 | 775 | | "Its outcome is unknown, so it is not re-run automatically. Size DurableFlowOptions.StateExpiry beyo |
| | 2 | 776 | | "or start a new parent run to re-execute the work."); |
| | | 777 | | } |
| | | 778 | | |
| | | 779 | | // Create the child BEFORE persisting the breadcrumb: "breadcrumb exists" must always |
| | | 780 | | // imply "child state existed", which keeps the expired-child check above sound. A crash |
| | | 781 | | // between the two writes is safe — the child id is deterministic, so the re-delivered |
| | | 782 | | // parent execution loads this child instead of re-creating it. |
| | 112 | 783 | | child = CreateChildState<TFlow, TInput>(childFlowId, name, inputJson); |
| | 112 | 784 | | if (await FlowStateConcurrency.TryCreateAsync( |
| | 112 | 785 | | _store, |
| | 112 | 786 | | childFlowId, |
| | 112 | 787 | | child, |
| | 112 | 788 | | _options.StateExpiry, |
| | 112 | 789 | | cancellationToken).ConfigureAwait(false)) |
| | | 790 | | { |
| | 112 | 791 | | _logger.LogDebug("Flow {FlowId} started child flow {ChildFlowId} for step '{Step}'.", FlowId, childFlowI |
| | | 792 | | } |
| | | 793 | | else |
| | | 794 | | { |
| | 0 | 795 | | child = await _store.LoadAsync(childFlowId, cancellationToken).ConfigureAwait(false) |
| | 0 | 796 | | ?? throw new InvalidOperationException($"Child flow '{childFlowId}' was created concurrently but cou |
| | 0 | 797 | | ThrowIfChildMismatched<TFlow, TInput>(child, childFlowId, name, inputJson); |
| | | 798 | | } |
| | | 799 | | } |
| | | 800 | | else |
| | | 801 | | { |
| | 128 | 802 | | ThrowIfChildMismatched<TFlow, TInput>(child, childFlowId, name, inputJson); |
| | | 803 | | } |
| | | 804 | | |
| | 228 | 805 | | if (breadcrumb is null) |
| | | 806 | | { |
| | 116 | 807 | | checkpoint.ChildFlowId = childFlowId; |
| | 116 | 808 | | checkpoint.Faulted = false; |
| | 116 | 809 | | checkpoint.Message = $"Waiting for child flow '{childFlowId}'."; |
| | 116 | 810 | | await SaveAsync(cancellationToken).ConfigureAwait(false); |
| | | 811 | | } |
| | | 812 | | |
| | 228 | 813 | | switch (child.Status) |
| | | 814 | | { |
| | | 815 | | // A terminal child snapshot is a settled outcome: memoize it uninterruptibly (local |
| | | 816 | | // and awaited-step parity) so a cancellation here cannot trip MarkLost on a healthy lease. |
| | | 817 | | // The caller gets the SNAPSHOT — the reduced shape the memo holds (no ambient Context, |
| | | 818 | | // nested child-step results elided) — on the first completion exactly as on every |
| | | 819 | | // replay, which reads it back from the memo above. Returning the loaded child here |
| | | 820 | | // handed the first execution a richer object than any re-execution would ever see, so |
| | | 821 | | // parent logic could branch differently (or fail) after a restart on a step it had |
| | | 822 | | // already completed; the whole point of the memo is that the two are indistinguishable. |
| | | 823 | | case FlowRunStatus.Succeeded: |
| | | 824 | | { |
| | 104 | 825 | | var snapshotJson = FlowStateJson.SerializeSnapshot(child); |
| | 104 | 826 | | await CompleteStepAsync(name, checkpoint, snapshotJson, CancellationToken.None, kind: DurableFlowStepKin |
| | 104 | 827 | | return MaterializeChildSnapshot(name, snapshotJson); |
| | | 828 | | } |
| | | 829 | | |
| | | 830 | | case FlowRunStatus.Failed: |
| | | 831 | | { |
| | 6 | 832 | | checkpoint.Message = child.LastMessage; |
| | 6 | 833 | | var snapshotJson = FlowStateJson.SerializeSnapshot(child); |
| | 6 | 834 | | await CompleteStepAsync(name, checkpoint, snapshotJson, CancellationToken.None, faulted: true, kind: Dur |
| | 6 | 835 | | var snapshot = MaterializeChildSnapshot(name, snapshotJson); |
| | 6 | 836 | | ThrowIfChildFailed(snapshot, failOnChildFailure); |
| | 4 | 837 | | return snapshot; |
| | | 838 | | } |
| | | 839 | | |
| | | 840 | | default: |
| | 310 | 841 | | await NotifyStepAsync(static (o, e) => o.OnStepWaitingAsync(e), name, DurableFlowStepKind.ChildFlow).Con |
| | 118 | 842 | | await SuspendForChildAsync(childFlowId, child, cancellationToken).ConfigureAwait(false); |
| | 0 | 843 | | throw new InvalidOperationException("Unreachable."); |
| | | 844 | | } |
| | 158 | 845 | | } |
| | | 846 | | |
| | | 847 | | private async Task<TResponse> AwaitStepCoreAsync<TResponse>( |
| | | 848 | | string name, |
| | | 849 | | Func<string, Task> trigger, |
| | | 850 | | Func<TResponse, ValueTask<bool>>? until, |
| | | 851 | | TimeSpan? timeout, |
| | | 852 | | CancellationToken cancellationToken) where TResponse : IAsyncResponsePayload |
| | | 853 | | { |
| | 1621 | 854 | | ThrowIfSuspended(); |
| | 1621 | 855 | | ArgumentException.ThrowIfNullOrWhiteSpace(name); |
| | 1621 | 856 | | ArgumentNullException.ThrowIfNull(trigger); |
| | | 857 | | |
| | 1621 | 858 | | using var active = EnterStep(name); |
| | 1621 | 859 | | var checkpoint = GetStep(name); |
| | 1621 | 860 | | if (checkpoint.Completed) |
| | 90 | 861 | | return DeserializeResult<TResponse>(checkpoint.ResultJson); |
| | | 862 | | |
| | | 863 | | // Re-attach when a previous execution already triggered this step and died waiting; start |
| | | 864 | | // fresh when there is no breadcrumb or the last attempt faulted (steps are idempotent). |
| | 1531 | 865 | | var reattach = checkpoint.PendingCorrelationId is not null && !checkpoint.Faulted; |
| | 1531 | 866 | | var correlationId = reattach |
| | 1531 | 867 | | ? checkpoint.PendingCorrelationId! |
| | 1531 | 868 | | : AsyncResponseContext.GenerateCorrelationId(); |
| | 1531 | 869 | | var stepTimeout = timeout ?? _options.DefaultStepTimeout; |
| | | 870 | | // The window the ledger must outlive: the resolved step timeout, or — for a timeout-less |
| | | 871 | | // wait — the channel's declared default waiter timeout, which the channel arms on the |
| | | 872 | | // waiter below anyway. Null only when the channel declares nothing; such waits keep the |
| | | 873 | | // historical plain-TTL stamp and re-arm-in-full replays. |
| | 1531 | 874 | | var waitWindow = stepTimeout ?? _channelDefaultWaitTimeout; |
| | | 875 | | |
| | 1715 | 876 | | await NotifyStepAsync(static (o, e) => o.OnStepStartingAsync(e), name, DurableFlowStepKind.Awaited, correlationI |
| | | 877 | | |
| | 1527 | 878 | | if (reattach && checkpoint.AwaitDeadlineUtc is { } awaitDeadline) |
| | | 879 | | { |
| | | 880 | | // The deadline persisted at the FIRST arm is the step's fault clock across |
| | | 881 | | // executions: a replay arms the REMAINDER, never a fresh full window. Recomputing the |
| | | 882 | | // window per attempt let every redelivery inside the timeout restart both the fault |
| | | 883 | | // clock and the ledger TTL from zero — a remote system that never answers produced a |
| | | 884 | | // run that neither completed nor alarmed, kept alive indefinitely. |
| | 14 | 885 | | var remainingWindow = awaitDeadline - UtcNow; |
| | 14 | 886 | | if (remainingWindow <= TimeSpan.Zero) |
| | | 887 | | { |
| | | 888 | | // The window elapsed while no execution was live. Recovery may have consumed the |
| | | 889 | | // response and completed the step in that gap — prefer its checkpoint over a |
| | | 890 | | // fault (same authority argument as the post-registration short-circuit below). |
| | 4 | 891 | | if (await TryShortCircuitRecoveredCheckpointAsync(name, checkpoint).ConfigureAwait(false)) |
| | 0 | 892 | | return DeserializeResult<TResponse>(checkpoint.ResultJson); |
| | | 893 | | |
| | | 894 | | // Settle exactly like the live timeout the waiter would have produced: the fault |
| | | 895 | | // is recorded so the next execution restarts the step fresh, and the exception |
| | | 896 | | // propagates as retriable for the transport's bounded redelivery. |
| | 4 | 897 | | var timedOut = new TimeoutException( |
| | 4 | 898 | | $"Timed out waiting for response for correlationId {correlationId}: the await deadline {awaitDeadlin |
| | 4 | 899 | | "elapsed while no execution was live."); |
| | 4 | 900 | | checkpoint.Faulted = true; |
| | 4 | 901 | | checkpoint.Message = timedOut.Message; |
| | 4 | 902 | | await SaveAsync(CancellationToken.None, cause: timedOut).ConfigureAwait(false); |
| | 4 | 903 | | throw timedOut; |
| | | 904 | | } |
| | | 905 | | |
| | 10 | 906 | | stepTimeout = remainingWindow; |
| | 10 | 907 | | waitWindow = remainingWindow; |
| | | 908 | | } |
| | | 909 | | |
| | 1523 | 910 | | var waiter = await CreateWaiterAsync(correlationId, until, stepTimeout, name).ConfigureAwait(false); |
| | 1521 | 911 | | var triggerCompleted = reattach; |
| | 1521 | 912 | | var notifyCompletion = false; |
| | | 913 | | try |
| | | 914 | | { |
| | 1521 | 915 | | if (reattach && await TryShortCircuitRecoveredCheckpointAsync(name, checkpoint).ConfigureAwait(false)) |
| | | 916 | | { |
| | | 917 | | // Lost-subscriber recovery checkpointed this step between our state load and the |
| | | 918 | | // waiter registration. Its wake-up delivery will find OUR lease alive and ack as |
| | | 919 | | // a duplicate, so nothing would ever wake the parked wait — take the checkpointed |
| | | 920 | | // result now instead of waiting out the full step timeout for a response that was |
| | | 921 | | // already consumed. (Recovery always checkpoints BEFORE enqueueing its wake-up, |
| | | 922 | | // so a completed persisted checkpoint here is authoritative.) |
| | 2 | 923 | | return DeserializeResult<TResponse>(checkpoint.ResultJson); |
| | | 924 | | } |
| | | 925 | | |
| | 1519 | 926 | | if (!reattach) |
| | | 927 | | { |
| | | 928 | | // Persist the breadcrumb AFTER the registration exists and BEFORE the send: |
| | | 929 | | // "breadcrumb persisted" therefore implies "someone is listening", so a crash on |
| | | 930 | | // either side of the send re-attaches (or times out and restarts the idempotent |
| | | 931 | | // step) — never a lost run, never a double-send. |
| | 1505 | 932 | | checkpoint.PendingCorrelationId = correlationId; |
| | 1505 | 933 | | checkpoint.PendingPayloadTypeFullName = typeof(TResponse).FullName; |
| | 1505 | 934 | | checkpoint.Faulted = false; |
| | 1505 | 935 | | checkpoint.Message = null; |
| | | 936 | | // The fault clock survives crashes and redeliveries only through this stamp (see |
| | | 937 | | // the re-attach deadline branch above); null when the effective window is unknown, |
| | | 938 | | // and such waits keep the recompute-per-attempt behavior. |
| | 1505 | 939 | | checkpoint.AwaitDeadlineUtc = waitWindow is { } window ? UtcNow.Add(window) : null; |
| | | 940 | | // The ledger must outlive the wait it records, exactly as SaveForSleepAsync covers |
| | | 941 | | // a timer's sleep: with a wait window longer than StateExpiry, a plain-TTL stamp |
| | | 942 | | // expires the row (and with it the lease renewal's anchor) mid-wait — the lease is |
| | | 943 | | // marked lost against a row that no longer exists and the run is unrecoverable. A |
| | | 944 | | // wait whose window is unknown keeps the plain stamp: bounding open-ended idleness |
| | | 945 | | // is what StateExpiry is documented to do. |
| | 1505 | 946 | | if (waitWindow is { } armWindow) |
| | 1483 | 947 | | await SaveForSleepAsync(armWindow, cancellationToken).ConfigureAwait(false); |
| | | 948 | | else |
| | 22 | 949 | | await SaveAsync(cancellationToken).ConfigureAwait(false); |
| | | 950 | | |
| | 1503 | 951 | | await trigger(correlationId).ConfigureAwait(false); |
| | 1493 | 952 | | triggerCompleted = true; |
| | | 953 | | } |
| | | 954 | | else |
| | | 955 | | { |
| | | 956 | | // Replayed execution re-attaching to an in-flight wait: the executor's |
| | | 957 | | // unconditional per-attempt save reset the ledger TTL to StateExpiry, so a wait |
| | | 958 | | // window longer than StateExpiry would out-live its own ledger and strand the |
| | | 959 | | // run mid-wait — re-extend to cover the wait, exactly as the fresh path above |
| | | 960 | | // and the timer path's replay branch do. With a persisted deadline the window is |
| | | 961 | | // the REMAINDER (shrunk above); a legacy ledger without one re-extends (and |
| | | 962 | | // re-arms) the full window — its fault clock restarts, the pre-deadline behavior. |
| | | 963 | | // A window-less re-attach keeps the plain stamp the executor already wrote. |
| | 14 | 964 | | if (waitWindow is { } replayWindow) |
| | 12 | 965 | | await SaveForSleepAsync(replayWindow, cancellationToken).ConfigureAwait(false); |
| | | 966 | | |
| | 12 | 967 | | if (_logger.IsEnabled(LogLevel.Debug)) |
| | | 968 | | { |
| | 0 | 969 | | _logger.LogDebug( |
| | 0 | 970 | | "Flow {FlowId} step '{Step}' re-attaching to in-flight correlationId {CorrelationId}.", |
| | 0 | 971 | | FlowId, name, correlationId); |
| | | 972 | | } |
| | | 973 | | } |
| | | 974 | | |
| | 1677 | 975 | | await NotifyStepAsync(static (o, e) => o.OnStepWaitingAsync(e), name, DurableFlowStepKind.Awaited, correlati |
| | | 976 | | |
| | 1503 | 977 | | WarnIfWaitOutlivesInFlightCeiling(name, waitWindow); |
| | 1503 | 978 | | var response = await WaitForResponseAsync(waiter.ResponseTask, cancellationToken).ConfigureAwait(false); |
| | | 979 | | |
| | 1456 | 980 | | checkpoint.PendingCorrelationId = null; |
| | 1456 | 981 | | checkpoint.PendingPayloadTypeFullName = null; |
| | | 982 | | // Deliberately NOT the caller's token: once the response is claimed from the channel |
| | | 983 | | // it exists nowhere else, so the completion checkpoint must not be interruptible — a |
| | | 984 | | // cancellation here used to leave `pending` set with the response already consumed, |
| | | 985 | | // and the redelivered execution re-attached to a correlation id nothing could answer. |
| | 1456 | 986 | | await CompleteStepAsync(name, checkpoint, AsyncResponseJson.Serialize(response), CancellationToken.None, kin |
| | 1450 | 987 | | notifyCompletion = true; |
| | 1450 | 988 | | return response; |
| | | 989 | | } |
| | 27 | 990 | | catch (OperationCanceledException ex) when (triggerCompleted) |
| | | 991 | | { |
| | | 992 | | // SETTLE the handoff before deciding. A point-in-time IsCompletedSuccessfully check |
| | | 993 | | // raced the channel's dispatch: the response could win the task a moment after the |
| | | 994 | | // check, leaving a consumed response behind a still-pending ledger. Disposing the |
| | | 995 | | // waiter cancels its response task unless something already completed it (the channel |
| | | 996 | | // contract since the dispose-cancels fix), so after this await the task is TERMINAL |
| | | 997 | | // and the decision below is the race's single authoritative outcome. The finally's |
| | | 998 | | // second dispose is a no-op behind the subscription's cleanup latch. |
| | 23 | 999 | | await waiter.DisposeAsync().ConfigureAwait(false); |
| | | 1000 | | |
| | 23 | 1001 | | if (waiter.ResponseTask.IsCompletedSuccessfully) |
| | | 1002 | | { |
| | | 1003 | | // Delivery won the settlement: the channel claimed and acked that message — it |
| | | 1004 | | // exists nowhere else, and re-attaching to its consumed correlation id would park |
| | | 1005 | | // the run until the step timeout. The checkpoint therefore wins over the |
| | | 1006 | | // cancellation: persist the received payload and return it; the caller's token |
| | | 1007 | | // gets its say again at the next step boundary. |
| | 4 | 1008 | | var received = await SettleWonResponseAsync(name, checkpoint, waiter.ResponseTask.Result, correlationId, |
| | 2 | 1009 | | notifyCompletion = true; |
| | 2 | 1010 | | return received; |
| | | 1011 | | } |
| | | 1012 | | |
| | | 1013 | | // No response was won, so nothing is at risk: a lost lease is now just the takeover |
| | | 1014 | | // signal it always was. |
| | 19 | 1015 | | _lease.ThrowIfLost(ex); |
| | | 1016 | | |
| | 19 | 1017 | | if (waiter.ResponseTask.IsFaulted) |
| | | 1018 | | { |
| | | 1019 | | // The wait FAULTED — a throwing Until predicate (possibly between the catch |
| | | 1020 | | // filter and the settlement), or the disposal drain abandoning a wedged delivery |
| | | 1021 | | // as AsyncResponseIndeterminateDeliveryException. Either way the message may be |
| | | 1022 | | // consumed: restart the idempotent step fresh, exactly like the general fault |
| | | 1023 | | // path below. The checkpoint records the fault's own message (not the |
| | | 1024 | | // cancellation's) so the ledger says WHY the step restarts. |
| | 2 | 1025 | | var fault = waiter.ResponseTask.Exception?.GetBaseException(); |
| | 2 | 1026 | | checkpoint.Faulted = true; |
| | 2 | 1027 | | checkpoint.Message = fault?.Message ?? ex.Message; |
| | 2 | 1028 | | await SaveAsync(CancellationToken.None, cause: fault ?? ex).ConfigureAwait(false); |
| | 2 | 1029 | | throw; |
| | | 1030 | | } |
| | | 1031 | | |
| | | 1032 | | // Cancellation won the settlement (the task is now canceled; nothing was delivered). |
| | | 1033 | | // WAIT-SIDE cancellation is infrastructure, not a step verdict: the channel cancels |
| | | 1034 | | // in-flight waiters when it is disposed at host shutdown, and the caller's token |
| | | 1035 | | // means "stop this execution", not "the step failed" — the remote operation is still |
| | | 1036 | | // in flight. The persisted breadcrumb must survive untouched so the redelivered |
| | | 1037 | | // execution RE-ATTACHES to the same correlation id; marking the checkpoint faulted |
| | | 1038 | | // here turned every graceful shutdown mid-await into a fresh-correlation restart that |
| | | 1039 | | // re-sent the remote request. (A response that never arrives still faults via the |
| | | 1040 | | // step timeout.) |
| | | 1041 | | // |
| | | 1042 | | // The filter keeps this branch away from TRIGGER-thrown cancellation (an HttpClient |
| | | 1043 | | // timeout surfaces as TaskCanceledException): the request may never have left the |
| | | 1044 | | // process, so that case falls through to the fault path below and restarts fresh. |
| | 17 | 1045 | | throw; |
| | 0 | 1046 | | } |
| | 34 | 1047 | | catch (Exception ex) |
| | | 1048 | | { |
| | 34 | 1049 | | if (triggerCompleted) |
| | | 1050 | | { |
| | | 1051 | | // The remote request is in flight (or already answered). SETTLE the handoff |
| | | 1052 | | // before deciding, exactly as the cancellation branch does: after this await the |
| | | 1053 | | // response task is terminal and the decision below is authoritative. |
| | 22 | 1054 | | await waiter.DisposeAsync().ConfigureAwait(false); |
| | | 1055 | | |
| | 22 | 1056 | | if (waiter.ResponseTask.IsCompletedSuccessfully) |
| | | 1057 | | { |
| | | 1058 | | // The response was won. Task.WaitAsync hands back a completed task BEFORE it |
| | | 1059 | | // consults the token, so a lease lost in the same instant the response landed |
| | | 1060 | | // returns the payload and lands here (not in the cancellation branch) when the |
| | | 1061 | | // fenced completion save trips ThrowIfLost — and the clock-based check inside |
| | | 1062 | | // that save can trip on its own in the same window. Settled exactly as the |
| | | 1063 | | // cancellation branch settles it: without this the redelivered execution |
| | | 1064 | | // re-attached to a consumed correlation id, burned the step timeout, and |
| | | 1065 | | // re-sent the request. |
| | 6 | 1066 | | var received = await SettleWonResponseAsync(name, checkpoint, waiter.ResponseTask.Result, correlatio |
| | 0 | 1067 | | notifyCompletion = true; |
| | 0 | 1068 | | return received; |
| | | 1069 | | } |
| | | 1070 | | |
| | 16 | 1071 | | if (!waiter.ResponseTask.IsFaulted) |
| | | 1072 | | { |
| | | 1073 | | // Nothing was delivered and the wait itself did not fault: the throw came from |
| | | 1074 | | // OUTSIDE the wait (a step observer, the logger, the replay branch's ledger |
| | | 1075 | | // re-extension) while the remote request was already sent. Marking the step |
| | | 1076 | | // faulted here made the redelivered execution mint a fresh correlation id and |
| | | 1077 | | // send the request AGAIN — the double-send the breadcrumb exists to prevent, |
| | | 1078 | | // and worse than a real crash, which leaves the breadcrumb intact. Keep it: |
| | | 1079 | | // the next execution re-attaches, or the persisted deadline faults it. |
| | 4 | 1080 | | checkpoint.Message = ex.Message; |
| | 4 | 1081 | | await SaveAsync(CancellationToken.None, cause: ex).ConfigureAwait(false); |
| | 2 | 1082 | | throw; |
| | | 1083 | | } |
| | | 1084 | | } |
| | | 1085 | | |
| | | 1086 | | // Timeout, trigger failure (including trigger-thrown cancellation), or a faulted |
| | | 1087 | | // wait: record it so the next execution restarts this step fresh instead of |
| | | 1088 | | // re-attaching to a dead correlation id. The original failure rides along as `cause` |
| | | 1089 | | // so a rejected save cannot displace it. |
| | 24 | 1090 | | checkpoint.Faulted = true; |
| | 24 | 1091 | | checkpoint.Message = ex.Message; |
| | 24 | 1092 | | await SaveAsync(CancellationToken.None, cause: ex).ConfigureAwait(false); |
| | 22 | 1093 | | throw; |
| | 0 | 1094 | | } |
| | | 1095 | | finally |
| | | 1096 | | { |
| | 1509 | 1097 | | await waiter.DisposeAsync().ConfigureAwait(false); |
| | | 1098 | | // Notification is outside the response-settlement catches: an observer failure |
| | | 1099 | | // must end this attempt after its durable checkpoint, not checkpoint and notify twice. |
| | 1509 | 1100 | | if (notifyCompletion) |
| | 1566 | 1101 | | await NotifyStepAsync(static (o, e) => o.OnStepCompletedAsync(e), name, DurableFlowStepKind.Awaited, cor |
| | | 1102 | | } |
| | 1536 | 1103 | | } |
| | | 1104 | | |
| | | 1105 | | /// <summary> |
| | | 1106 | | /// An awaited step holds its delivery for the whole wait, and — unlike a timer — it is NOT |
| | | 1107 | | /// handed over to a fresh delivery at the in-process budget: disposing the waiter deletes its |
| | | 1108 | | /// lost-subscriber recovery registration, so a response landing between one hop's waiter and |
| | | 1109 | | /// the next hop's re-attach would find neither a subscriber nor a recovery target and be |
| | | 1110 | | /// dropped. On a transport with an in-flight ceiling a wait longer than the budget can |
| | | 1111 | | /// therefore outlive its delivery: the broker redelivers the job while this handler is still |
| | | 1112 | | /// parked, and the copy contends on the execution lease this handler holds. That is |
| | | 1113 | | /// configuration the operator can fix and should hear about — hence one warning per parked |
| | | 1114 | | /// step. |
| | | 1115 | | /// </summary> |
| | | 1116 | | private void WarnIfWaitOutlivesInFlightCeiling(string name, TimeSpan? waitWindow) |
| | | 1117 | | { |
| | 1503 | 1118 | | if (_workerTransport is not IWorkerTransportInFlightLimit { MaxInFlightDuration: { } ceiling } |
| | 1503 | 1119 | | || ceiling <= TimeSpan.Zero |
| | 1503 | 1120 | | || InProcessParkBudget() is not { } budget |
| | 1503 | 1121 | | || waitWindow <= budget) |
| | | 1122 | | { |
| | 1501 | 1123 | | return; |
| | | 1124 | | } |
| | | 1125 | | |
| | 2 | 1126 | | _logger.LogWarning( |
| | 2 | 1127 | | "Flow {FlowId} step '{Step}' waits in process for up to {WaitWindow} for its response, but the worker transp |
| | 2 | 1128 | | FlowId, |
| | 2 | 1129 | | name, |
| | 2 | 1130 | | waitWindow?.ToString() ?? "an unbounded time", |
| | 2 | 1131 | | ceiling, |
| | 2 | 1132 | | budget); |
| | 2 | 1133 | | } |
| | | 1134 | | |
| | | 1135 | | /// <summary> |
| | | 1136 | | /// Checkpoints a response that WON the waiter's settlement while the attempt was already |
| | | 1137 | | /// unwinding (a cancellation, or a throw from outside the wait). The one place this is done, |
| | | 1138 | | /// because both unwinding branches need it and each was once fixed without the other. |
| | | 1139 | | /// <para> |
| | | 1140 | | /// The lease is checked AFTER settling, never before. Running ThrowIfLost first threw while |
| | | 1141 | | /// the waiter still held a claimed, channel-acked response: the payload was dropped with no |
| | | 1142 | | /// checkpoint and no re-publish, PendingCorrelationId stayed set, and the redelivered |
| | | 1143 | | /// execution re-attached to a correlation id that could never be answered — one lost response |
| | | 1144 | | /// plus one duplicate remote request. A lost lease cannot write lease-fenced, so the payload |
| | | 1145 | | /// is persisted through the lease-less compare-and-swap the recovery path already uses; only |
| | | 1146 | | /// then is the takeover signal raised, with <paramref name="cause"/> attached. |
| | | 1147 | | /// </para> |
| | | 1148 | | /// </summary> |
| | | 1149 | | private async Task<TResponse> SettleWonResponseAsync<TResponse>( |
| | | 1150 | | string name, |
| | | 1151 | | FlowStepState checkpoint, |
| | | 1152 | | TResponse received, |
| | | 1153 | | string correlationId, |
| | | 1154 | | Exception cause) |
| | | 1155 | | { |
| | 10 | 1156 | | checkpoint.PendingCorrelationId = null; |
| | 10 | 1157 | | if (_lease.IsLost) |
| | | 1158 | | { |
| | 8 | 1159 | | await CheckpointReceivedWithoutLeaseAsync(name, checkpoint, received, correlationId).ConfigureAwait(false); |
| | 8 | 1160 | | _lease.ThrowIfLost(cause); |
| | | 1161 | | } |
| | | 1162 | | |
| | 2 | 1163 | | await CompleteStepAsync(name, checkpoint, AsyncResponseJson.Serialize(received), CancellationToken.None, kind: D |
| | 2 | 1164 | | return received; |
| | 2 | 1165 | | } |
| | | 1166 | | |
| | | 1167 | | /// <summary> |
| | | 1168 | | /// Re-reads the persisted step checkpoint after the re-attach waiter registration exists and, |
| | | 1169 | | /// when recovery already completed the step, syncs the in-memory ledger so the caller can |
| | | 1170 | | /// short-circuit. Best-effort: a store read failure logs and falls through to the normal wait |
| | | 1171 | | /// (the behavior before this check existed) rather than faulting the step. |
| | | 1172 | | /// </summary> |
| | | 1173 | | private async Task<bool> TryShortCircuitRecoveredCheckpointAsync(string name, FlowStepState checkpoint) |
| | | 1174 | | { |
| | | 1175 | | try |
| | | 1176 | | { |
| | 20 | 1177 | | var persisted = await _store.LoadAsync(FlowId).ConfigureAwait(false); |
| | 20 | 1178 | | if (persisted?.Steps is null |
| | 20 | 1179 | | || !persisted.Steps.TryGetValue(name, out var persistedStep) |
| | 20 | 1180 | | || !persistedStep.Completed) |
| | | 1181 | | { |
| | 16 | 1182 | | return false; |
| | | 1183 | | } |
| | | 1184 | | |
| | | 1185 | | // Adopt ONLY while the persisted run is still Running. The revision sync below makes |
| | | 1186 | | // the next checkpoint's CAS succeed, so adopting the revision of a writer that ALSO |
| | | 1187 | | // transitioned the run (RecoverAsync's escalation into FailAsync marking it Failed, an |
| | | 1188 | | // operator parking it) would let this execution's next save write the stale in-memory |
| | | 1189 | | // Status/LastMessage over that transition — resurrecting a terminally Failed run. |
| | | 1190 | | // Falling through keeps the stale revision, so the next save loses the CAS and the |
| | | 1191 | | // delivery abandons and retries: the documented outcome for losing a concurrent write. |
| | 4 | 1192 | | if (persisted.Status is not FlowRunStatus.Running) |
| | 2 | 1193 | | return false; |
| | | 1194 | | |
| | | 1195 | | // Sync the ledger revision too: the recovery write that completed this step advanced |
| | | 1196 | | // it, and a stale in-memory revision would fail the NEXT checkpoint's compare-and-swap |
| | | 1197 | | // — aborting every execution that took this short-circuit as a phantom "concurrent |
| | | 1198 | | // write" and forcing a pointless redelivery. |
| | 2 | 1199 | | _state.Revision = persisted.Revision; |
| | 2 | 1200 | | checkpoint.Completed = true; |
| | 2 | 1201 | | checkpoint.ResultJson = persistedStep.ResultJson; |
| | 2 | 1202 | | checkpoint.PendingCorrelationId = null; |
| | 2 | 1203 | | checkpoint.PendingPayloadTypeFullName = null; |
| | 2 | 1204 | | checkpoint.Faulted = false; |
| | 2 | 1205 | | checkpoint.Message = persistedStep.Message; |
| | 2 | 1206 | | checkpoint.CompletedAtUtc = persistedStep.CompletedAtUtc; |
| | 2 | 1207 | | MarkStepReturned(name); |
| | 2 | 1208 | | return true; |
| | | 1209 | | } |
| | 0 | 1210 | | catch (Exception ex) |
| | | 1211 | | { |
| | 0 | 1212 | | _logger.LogWarning( |
| | 0 | 1213 | | ex, |
| | 0 | 1214 | | "Flow {FlowId} step '{Step}' could not re-read its checkpoint before re-attaching; continuing with the n |
| | 0 | 1215 | | FlowId, name); |
| | 0 | 1216 | | return false; |
| | | 1217 | | } |
| | 20 | 1218 | | } |
| | | 1219 | | |
| | | 1220 | | private async Task<IAsyncResponseWaiter<TResponse>> CreateWaiterAsync<TResponse>( |
| | | 1221 | | string correlationId, |
| | | 1222 | | Func<TResponse, ValueTask<bool>>? until, |
| | | 1223 | | TimeSpan? timeout, |
| | | 1224 | | string stepName) where TResponse : IAsyncResponsePayload |
| | | 1225 | | { |
| | 1523 | 1226 | | if (_recoverableSubscriber is not null) |
| | | 1227 | | { |
| | | 1228 | | // The durable safety net: a response landing while no process is executing this flow |
| | | 1229 | | // checkpoints the terminal payload and re-enqueues the run, or terminally fails it — |
| | | 1230 | | // the same at-least-once, idempotency-required contract as hand-registered callbacks. |
| | 1491 | 1231 | | var flowId = FlowId; |
| | 1491 | 1232 | | Expression<Func<IDurableFlowExecutor, Task>> resume = executor => executor.RecoverAsync( |
| | 1491 | 1233 | | flowId, |
| | 1491 | 1234 | | Placeholder.Payload<TResponse>()!, |
| | 1491 | 1235 | | Placeholder.CorrelationId()); |
| | | 1236 | | // Correlation-scoped like the resume target: a dead worker's registration outlives |
| | | 1237 | | // the replacement's, so an unscoped failure let a late error for a superseded |
| | | 1238 | | // correlation id terminally fail a run that was live on another one. |
| | 1491 | 1239 | | Expression<Func<IDurableFlowExecutor, Task>> failure = executor => executor.FailAsync( |
| | 1491 | 1240 | | flowId, |
| | 1491 | 1241 | | Placeholder.Exception(), |
| | 1491 | 1242 | | Placeholder.CorrelationId()); |
| | | 1243 | | |
| | 1491 | 1244 | | return await _recoverableSubscriber.CreateRecoverableResponseWaiter( |
| | 1491 | 1245 | | correlationId, |
| | 1491 | 1246 | | CallbackExpressionConverter.ToReflectionCall(resume), |
| | 1491 | 1247 | | CallbackExpressionConverter.ToReflectionCall(failure), |
| | 1491 | 1248 | | until, |
| | 1491 | 1249 | | timeout).ConfigureAwait(false); |
| | | 1250 | | } |
| | | 1251 | | |
| | 32 | 1252 | | _logger.LogDebug( |
| | 32 | 1253 | | "Flow {FlowId} step '{Step}': the configured channel exposes no recoverable subscriber; lost-subscriber reco |
| | 32 | 1254 | | FlowId, stepName); |
| | | 1255 | | |
| | 32 | 1256 | | return await _subscriber.CreateResponseWaiter(correlationId, until, timeout).ConfigureAwait(false); |
| | 1521 | 1257 | | } |
| | | 1258 | | |
| | | 1259 | | private FlowState CreateChildState<TFlow, TInput>(string flowId, string parentStepName, string inputJson) |
| | | 1260 | | { |
| | 112 | 1261 | | var now = UtcNow; |
| | 112 | 1262 | | return new FlowState |
| | 112 | 1263 | | { |
| | 112 | 1264 | | FlowId = flowId, |
| | 112 | 1265 | | FlowTypeName = typeof(TFlow).FullName, |
| | 112 | 1266 | | InputTypeName = typeof(TInput).FullName, |
| | 112 | 1267 | | InputJson = inputJson, |
| | 112 | 1268 | | Status = FlowRunStatus.Running, |
| | 112 | 1269 | | LastMessage = $"Child flow started by {FlowId}.", |
| | 112 | 1270 | | CreatedAtUtc = now, |
| | 112 | 1271 | | UpdatedAtUtc = now, |
| | 112 | 1272 | | ParentFlowId = FlowId, |
| | 112 | 1273 | | ParentStepName = parentStepName, |
| | 112 | 1274 | | Context = _propagation.Capture() |
| | 112 | 1275 | | }; |
| | | 1276 | | } |
| | | 1277 | | |
| | | 1278 | | private Task EnqueueChildAsync(string childFlowId) |
| | | 1279 | | { |
| | 118 | 1280 | | var id = childFlowId; |
| | 118 | 1281 | | return _builder.EnqueueWorkerAsync<IDurableFlowExecutor>(executor => executor.ExecuteAsync(id)); |
| | | 1282 | | } |
| | | 1283 | | |
| | | 1284 | | private async Task SuspendForChildAsync(string childFlowId, FlowState? child, CancellationToken cancellationToken) |
| | | 1285 | | { |
| | | 1286 | | // Persist the suspension BEFORE the child becomes runnable: once the child is enqueued it |
| | | 1287 | | // can complete and re-execute this parent on another worker at any moment, and a save after |
| | | 1288 | | // that point would clobber the re-execution's newer checkpoints with this stale snapshot. |
| | | 1289 | | // The executor therefore does NOT save again on the suspension path. |
| | 118 | 1290 | | _state.LastMessage = $"Flow {FlowId} suspended waiting for child flow {childFlowId}."; |
| | | 1291 | | |
| | | 1292 | | // Cover the child's OWN park window, not just this parent's idle margin. A plain |
| | | 1293 | | // StateExpiry save here (and the executor's per-attempt save above it) SHRANK a ledger the |
| | | 1294 | | // child had already extended through ExtendAncestorLedgersAsync — and nothing re-extends |
| | | 1295 | | // it while the child is parked in-process under a live lease, because that rescue enqueue |
| | | 1296 | | // is acked as redundant and the child never replays. The parent's row then expired |
| | | 1297 | | // mid-park and the child's completion wake-up found no state: the parent run, and every |
| | | 1298 | | // step after this one, silently lost. |
| | 236 | 1299 | | await ParkAsync(RemainingChildParkWindow(child), () => EnqueueChildAsync(childFlowId), cancellationToken).Config |
| | 0 | 1300 | | } |
| | | 1301 | | |
| | | 1302 | | /// <summary> |
| | | 1303 | | /// How long the child's own persisted park runs from here — its pending timer wake or awaited |
| | | 1304 | | /// deadline, whichever is furthest out. Zero when the child is simply running, which leaves |
| | | 1305 | | /// the plain StateExpiry behavior unchanged. |
| | | 1306 | | /// </summary> |
| | | 1307 | | private TimeSpan RemainingChildParkWindow(FlowState? child) |
| | | 1308 | | { |
| | 118 | 1309 | | if (child is null) |
| | 0 | 1310 | | return TimeSpan.Zero; |
| | | 1311 | | |
| | 118 | 1312 | | if (child.Steps is not { Count: > 0 } steps) |
| | 116 | 1313 | | return TimeSpan.Zero; |
| | | 1314 | | |
| | 2 | 1315 | | var now = UtcNow; |
| | 2 | 1316 | | var furthest = now; |
| | 8 | 1317 | | foreach (var step in steps.Values) |
| | | 1318 | | { |
| | 2 | 1319 | | if (step.Completed) |
| | | 1320 | | continue; |
| | | 1321 | | |
| | 2 | 1322 | | if (step.WakeAtUtc is { } wakeAt && wakeAt > furthest) |
| | 2 | 1323 | | furthest = wakeAt; |
| | | 1324 | | |
| | 2 | 1325 | | if (step.AwaitDeadlineUtc is { } deadline && deadline > furthest) |
| | 0 | 1326 | | furthest = deadline; |
| | | 1327 | | } |
| | | 1328 | | |
| | 2 | 1329 | | return furthest > now ? furthest - now : TimeSpan.Zero; |
| | | 1330 | | } |
| | | 1331 | | |
| | | 1332 | | private void MarkStepReturned(string name) |
| | 5517 | 1333 | | => (_returnedSteps ??= new HashSet<string>(StringComparer.Ordinal)).Add(name); |
| | | 1334 | | |
| | | 1335 | | /// <summary> |
| | | 1336 | | /// Marks a step call in flight for its whole duration. A flow body is sequential by contract |
| | | 1337 | | /// and nothing here is thread-safe: two steps running at once (<c>Task.WhenAll</c> over two |
| | | 1338 | | /// context calls) interleave their writes of one ledger and one revision counter, which used to |
| | | 1339 | | /// surface — sometimes — as a rejected checkpoint blamed on a lost execution lease. A second |
| | | 1340 | | /// call that starts while one is in flight fails immediately with the actual reason instead. |
| | | 1341 | | /// A step called from INSIDE the running step's own body is sequential and stays allowed. |
| | | 1342 | | /// </summary> |
| | | 1343 | | private StepScope EnterStep(string name) |
| | | 1344 | | { |
| | 5832 | 1345 | | if (Interlocked.CompareExchange(ref _activeStep, 1, 0) == 0) |
| | | 1346 | | { |
| | | 1347 | | // Scoped to the calling step method: an async method's changes to the execution |
| | | 1348 | | // context never flow back to its caller, so the flow body itself never sees this. |
| | 5832 | 1349 | | ActiveStepOwner.Value = _stepToken; |
| | 5832 | 1350 | | return new StepScope(this); |
| | | 1351 | | } |
| | | 1352 | | |
| | 0 | 1353 | | if (ReferenceEquals(ActiveStepOwner.Value, _stepToken)) |
| | 0 | 1354 | | return default; |
| | | 1355 | | |
| | 0 | 1356 | | throw new InvalidOperationException( |
| | 0 | 1357 | | $"Step '{name}' of flow '{FlowId}' was started while another step of the same run was still executing. A dur |
| | 0 | 1358 | | "its steps sequentially — await each context call before making the next one (no Task.WhenAll over steps). F |
| | 0 | 1359 | | "start child flows or run the parallel part inside one step."); |
| | | 1360 | | } |
| | | 1361 | | |
| | | 1362 | | private readonly struct StepScope(DurableFlowContext? owner) : IDisposable |
| | | 1363 | | { |
| | | 1364 | | public void Dispose() |
| | | 1365 | | { |
| | 5818 | 1366 | | if (owner is not null) |
| | 5818 | 1367 | | Volatile.Write(ref owner._activeStep, 0); |
| | 5818 | 1368 | | } |
| | | 1369 | | } |
| | | 1370 | | |
| | | 1371 | | private void ThrowIfSuspended() |
| | | 1372 | | { |
| | 7282 | 1373 | | _lease.ThrowIfLost(); |
| | 7282 | 1374 | | _parkFailure?.Throw(); |
| | 7280 | 1375 | | if (_suspended) |
| | 0 | 1376 | | throw new DurableFlowSuspendedException(_state.LastMessage ?? $"Flow {FlowId} is suspended."); |
| | 7280 | 1377 | | } |
| | | 1378 | | |
| | | 1379 | | /// <summary> |
| | | 1380 | | /// Reads a just-memoized child snapshot back through the SAME deserializer the replay branch |
| | | 1381 | | /// uses, so the object handed to the first completion is bit-for-bit what every later |
| | | 1382 | | /// execution receives. |
| | | 1383 | | /// </summary> |
| | | 1384 | | private FlowState MaterializeChildSnapshot(string stepName, string snapshotJson) |
| | 110 | 1385 | | => DeserializeResult<FlowState>(snapshotJson) |
| | 110 | 1386 | | ?? throw new DurableFlowFailedException( |
| | 110 | 1387 | | $"Completed child step '{stepName}' of flow '{FlowId}' has no child-state snapshot."); |
| | | 1388 | | |
| | | 1389 | | private static void ThrowIfChildFailed(FlowState child, bool failOnChildFailure) |
| | | 1390 | | { |
| | 56 | 1391 | | if (failOnChildFailure && child.Status == FlowRunStatus.Failed) |
| | 2 | 1392 | | throw new DurableFlowFailedException($"Child flow '{child.FlowId}' failed: {child.LastMessage ?? "no message |
| | 54 | 1393 | | } |
| | | 1394 | | |
| | | 1395 | | private void ThrowIfChildMismatched<TFlow, TInput>( |
| | | 1396 | | FlowState child, |
| | | 1397 | | string childFlowId, |
| | | 1398 | | string stepName, |
| | | 1399 | | string requestedInputJson, |
| | | 1400 | | bool completed = false) |
| | | 1401 | | { |
| | | 1402 | | // A child id is owned by exactly one parent: the notification that resumes a suspended |
| | | 1403 | | // parent follows the child's single ParentFlowId, so a second parent awaiting the same id |
| | | 1404 | | // would suspend and never wake. Reject collisions loudly instead of parking forever. |
| | 178 | 1405 | | if (!string.Equals(child.ParentFlowId, FlowId, StringComparison.Ordinal)) |
| | | 1406 | | { |
| | 2 | 1407 | | var owner = child.ParentFlowId is null ? "a run not started by AwaitChildFlowAsync" : $"parent flow '{child. |
| | 2 | 1408 | | throw new DurableFlowFailedException( |
| | 2 | 1409 | | $"Step '{stepName}' of flow '{FlowId}' awaits child flow id '{childFlowId}', but that id belongs to {own |
| | 2 | 1410 | | "Child flow ids are exclusive to the parent that started them — pass a flowId that is unique per parent |
| | 2 | 1411 | | "(the default '{parentFlowId}:{stepName}' id is always safe)."); |
| | | 1412 | | } |
| | | 1413 | | |
| | 176 | 1414 | | if (!string.Equals(child.FlowId, childFlowId, StringComparison.Ordinal) |
| | 176 | 1415 | | || !string.Equals(child.ParentStepName, stepName, StringComparison.Ordinal)) |
| | | 1416 | | { |
| | 2 | 1417 | | throw new DurableFlowFailedException( |
| | 2 | 1418 | | $"Child flow id '{childFlowId}' is bound to a different child step than '{stepName}' of parent flow '{Fl |
| | 2 | 1419 | | "A child id is exclusive to one parent step."); |
| | | 1420 | | } |
| | | 1421 | | |
| | 174 | 1422 | | if (!string.Equals(child.FlowTypeName, typeof(TFlow).FullName, StringComparison.Ordinal)) |
| | | 1423 | | { |
| | 2 | 1424 | | throw new DurableFlowFailedException( |
| | 2 | 1425 | | $"Step '{stepName}' of flow '{FlowId}' awaits child flow id '{childFlowId}' as {typeof(TFlow).FullName}, |
| | 2 | 1426 | | $"but the persisted run is {child.FlowTypeName}. The flowId collides with a different flow — use a uniqu |
| | | 1427 | | } |
| | | 1428 | | |
| | | 1429 | | // The VALUE is compared, not the JSON shape the serializer happened to give it when the |
| | | 1430 | | // child was created: a member added to TInput since (nulls and defaults are written) made |
| | | 1431 | | // every in-flight parent's replay differ from its own persisted child and fail terminally. |
| | 172 | 1432 | | if (string.Equals(child.InputTypeName, typeof(TInput).FullName, StringComparison.Ordinal) |
| | 172 | 1433 | | && FlowStateJson.InputEquivalent<TInput>(child.InputJson, requestedInputJson)) |
| | | 1434 | | { |
| | 164 | 1435 | | return; |
| | | 1436 | | } |
| | | 1437 | | |
| | 8 | 1438 | | if (completed) |
| | | 1439 | | { |
| | | 1440 | | // A completed step answers from its memo whatever the current arguments are — a |
| | | 1441 | | // local step never re-reads its lambda, a timer never re-reads its delay. The child |
| | | 1442 | | // finished (possibly weeks ago) and its outcome is settled; failing the PARENT |
| | | 1443 | | // terminally over an input edit made since would throw that outcome away. |
| | 2 | 1444 | | _logger.LogWarning( |
| | 2 | 1445 | | "Flow {FlowId} step '{Step}' requested child flow {ChildFlowId} with a different input type or value tha |
| | 2 | 1446 | | FlowId, stepName, childFlowId); |
| | 2 | 1447 | | return; |
| | | 1448 | | } |
| | | 1449 | | |
| | 6 | 1450 | | throw new DurableFlowFailedException( |
| | 6 | 1451 | | $"Step '{stepName}' of flow '{FlowId}' requested child flow id '{childFlowId}' with a different input " + |
| | 6 | 1452 | | "type or value than the persisted child. Replays must use semantically identical child input."); |
| | | 1453 | | } |
| | | 1454 | | |
| | | 1455 | | private FlowStepState GetStep(string name) |
| | | 1456 | | { |
| | | 1457 | | // A name that already RETURNED in this execution is being used for a second step. The |
| | | 1458 | | // checkpoint is keyed by name alone, so the second use would be answered from the first |
| | | 1459 | | // one's memo: a step inside a loop ran its first iteration and silently skipped the rest, |
| | | 1460 | | // returning iteration one's result every time. A step that THREW is not recorded, so |
| | | 1461 | | // retrying it under its name within one execution keeps working. |
| | 5848 | 1462 | | if (_returnedSteps is not null && _returnedSteps.Contains(name)) |
| | | 1463 | | { |
| | 0 | 1464 | | throw new InvalidOperationException( |
| | 0 | 1465 | | $"Step name '{name}' was already used in this execution of flow '{FlowId}'. Checkpoints are keyed by ste |
| | 0 | 1466 | | "step with the same name would be skipped and handed the first one's result. Give every step a unique na |
| | 0 | 1467 | | "put the iteration key in it (for example $\"send-{item.Id}\")."); |
| | | 1468 | | } |
| | | 1469 | | |
| | 5848 | 1470 | | var steps = _state.Steps ??= new Dictionary<string, FlowStepState>(StringComparer.Ordinal); |
| | 5848 | 1471 | | if (!steps.TryGetValue(name, out var step)) |
| | | 1472 | | { |
| | 5370 | 1473 | | if (_options.MaxRetainedSteps is { } limit && steps.Count >= limit) |
| | 4 | 1474 | | throw new DurableFlowFailedException( |
| | 4 | 1475 | | $"Flow '{FlowId}' cannot add step '{name}': its {limit}-step MaxRetainedSteps budget is exhausted. " |
| | 4 | 1476 | | "No side effects of this step were started. Partition the work into bounded child flows, or explicit |
| | 5366 | 1477 | | step = new FlowStepState(); |
| | 5366 | 1478 | | steps[name] = step; |
| | | 1479 | | } |
| | 478 | 1480 | | else if (step.Completed) |
| | | 1481 | | { |
| | | 1482 | | // Every caller returns a completed step's memo straight away. |
| | 236 | 1483 | | MarkStepReturned(name); |
| | | 1484 | | } |
| | | 1485 | | |
| | 5844 | 1486 | | return step; |
| | | 1487 | | } |
| | | 1488 | | |
| | | 1489 | | private async Task CompleteStepAsync( |
| | | 1490 | | string name, |
| | | 1491 | | FlowStepState step, |
| | | 1492 | | string? resultJson, |
| | | 1493 | | CancellationToken cancellationToken, |
| | | 1494 | | bool faulted = false, |
| | | 1495 | | DurableFlowStepKind kind = DurableFlowStepKind.Local, |
| | | 1496 | | string? correlationId = null, |
| | | 1497 | | bool notify = true) |
| | | 1498 | | { |
| | 5279 | 1499 | | step.Completed = true; |
| | 5279 | 1500 | | step.ResultJson = resultJson; |
| | 5279 | 1501 | | step.PendingCorrelationId = null; |
| | | 1502 | | // Cleared together with the breadcrumb on EVERY settlement path (the FlowState contract): |
| | | 1503 | | // a stale declared-type name on a completed step would mislead the next recovery pass. |
| | 5279 | 1504 | | step.PendingPayloadTypeFullName = null; |
| | | 1505 | | // A memoized failed child keeps Faulted = true so operators can spot the failure on the |
| | | 1506 | | // step itself instead of digging through ResultJson. |
| | 5279 | 1507 | | step.Faulted = faulted; |
| | 5279 | 1508 | | step.CompletedAtUtc = UtcNow; |
| | 5279 | 1509 | | _state.LastMessage = faulted ? $"Step '{name}' completed (child flow failed)." : $"Step '{name}' completed."; |
| | 5279 | 1510 | | MarkStepReturned(name); |
| | 5279 | 1511 | | await SaveAsync(cancellationToken).ConfigureAwait(false); |
| | | 1512 | | |
| | 5267 | 1513 | | if (_logger.IsEnabled(LogLevel.Debug)) |
| | 18 | 1514 | | _logger.LogDebug("Flow {FlowId} step '{Step}' completed.", FlowId, name); |
| | | 1515 | | |
| | 5267 | 1516 | | if (notify) |
| | 7153 | 1517 | | await NotifyStepAsync(static (o, e) => o.OnStepCompletedAsync(e), name, kind, correlationId, step.WakeAtUtc) |
| | 5261 | 1518 | | } |
| | | 1519 | | |
| | | 1520 | | /// <summary> |
| | | 1521 | | /// Persists a response that was already claimed and acked by the channel when this execution's |
| | | 1522 | | /// lease had ALREADY been lost — the one case where a lease-fenced write is impossible but the |
| | | 1523 | | /// payload exists nowhere else. Uses the same lease-less compare-and-swap the recovery |
| | | 1524 | | /// dispatcher uses, and re-reads the ledger so it mutates whatever the new owner wrote rather |
| | | 1525 | | /// than clobbering it. Best-effort by construction: on a conflict or an absent ledger the step |
| | | 1526 | | /// simply restarts, which is the pre-existing behavior — but on the common path the response |
| | | 1527 | | /// survives instead of being dropped. |
| | | 1528 | | /// <para> |
| | | 1529 | | /// Fenced to THIS attempt, exactly as <c>DurableFlowExecutor.RecoverAsync</c> fences a recovered |
| | | 1530 | | /// payload: the reloaded step must still be pending on <paramref name="correlationId"/> and the |
| | | 1531 | | /// run must still be checkpointable. Losing the lease means a takeover may already have run — |
| | | 1532 | | /// timed the breadcrumb out, re-triggered the step under a NEW correlation id, or failed the |
| | | 1533 | | /// run — and a write keyed only on "step name, not completed" would complete the newer |
| | | 1534 | | /// attempt's pending step with this attempt's stale response (revision CAS cannot catch it: |
| | | 1535 | | /// the mutation deliberately targets the freshly loaded revision). A stale response is |
| | | 1536 | | /// discarded with a warning; the newer attempt's own response is the one that counts. |
| | | 1537 | | /// </para> |
| | | 1538 | | /// </summary> |
| | | 1539 | | private async Task CheckpointReceivedWithoutLeaseAsync<T>( |
| | | 1540 | | string name, |
| | | 1541 | | FlowStepState step, |
| | | 1542 | | T received, |
| | | 1543 | | string correlationId) |
| | | 1544 | | { |
| | 8 | 1545 | | var resultJson = AsyncResponseJson.Serialize(received); |
| | 8 | 1546 | | var completedAtUtc = UtcNow; |
| | 8 | 1547 | | var applied = false; |
| | 8 | 1548 | | string? skipReason = null; |
| | | 1549 | | |
| | | 1550 | | try |
| | | 1551 | | { |
| | 8 | 1552 | | var found = await FlowStateConcurrency.MutateAsync( |
| | 8 | 1553 | | _store, |
| | 8 | 1554 | | FlowId, |
| | 8 | 1555 | | _options.StateExpiry, |
| | 8 | 1556 | | _timeProvider, |
| | 8 | 1557 | | state => |
| | 8 | 1558 | | { |
| | 8 | 1559 | | applied = false; |
| | 8 | 1560 | | skipReason = null; |
| | 8 | 1561 | | |
| | 8 | 1562 | | // Same eligibility as RecoverAsync: Suspended runs still take the checkpoint |
| | 8 | 1563 | | // (an operator parked the run; the payload exists nowhere else and un-parking |
| | 8 | 1564 | | // replays from it), terminal runs never do. |
| | 8 | 1565 | | if (state.Status is not (FlowRunStatus.Running or FlowRunStatus.Suspended)) |
| | 8 | 1566 | | { |
| | 2 | 1567 | | skipReason = $"the run is {state.Status}"; |
| | 2 | 1568 | | return false; |
| | 8 | 1569 | | } |
| | 8 | 1570 | | |
| | 6 | 1571 | | if (state.Steps is not { } steps || !steps.TryGetValue(name, out var current)) |
| | 8 | 1572 | | { |
| | 0 | 1573 | | skipReason = "the step no longer exists in the ledger"; |
| | 0 | 1574 | | return false; |
| | 8 | 1575 | | } |
| | 8 | 1576 | | |
| | 6 | 1577 | | if (current.Completed) |
| | 8 | 1578 | | { |
| | 0 | 1579 | | skipReason = "the step is already completed"; |
| | 0 | 1580 | | return false; |
| | 8 | 1581 | | } |
| | 8 | 1582 | | |
| | 6 | 1583 | | if (!string.Equals(current.PendingCorrelationId, correlationId, StringComparison.Ordinal)) |
| | 8 | 1584 | | { |
| | 2 | 1585 | | skipReason = current.PendingCorrelationId is null |
| | 2 | 1586 | | ? "the step is no longer pending on any correlation id" |
| | 2 | 1587 | | : "the step is pending on a newer correlation id (a takeover re-triggered it)"; |
| | 2 | 1588 | | return false; |
| | 8 | 1589 | | } |
| | 8 | 1590 | | |
| | 4 | 1591 | | current.Completed = true; |
| | 4 | 1592 | | current.ResultJson = resultJson; |
| | 4 | 1593 | | current.PendingCorrelationId = null; |
| | 4 | 1594 | | current.PendingPayloadTypeFullName = null; |
| | 4 | 1595 | | current.Faulted = false; |
| | 4 | 1596 | | current.CompletedAtUtc = completedAtUtc; |
| | 4 | 1597 | | state.LastMessage = $"Step '{name}' completed (checkpointed after the execution lease was lost)."; |
| | 4 | 1598 | | applied = true; |
| | 4 | 1599 | | return true; |
| | 8 | 1600 | | }, |
| | 8 | 1601 | | CancellationToken.None).ConfigureAwait(false); |
| | | 1602 | | |
| | 8 | 1603 | | if (applied) |
| | | 1604 | | { |
| | 4 | 1605 | | _logger.LogWarning( |
| | 4 | 1606 | | "Flow {FlowId} lost its execution lease while step '{Step}' held a claimed response for correlationI |
| | 4 | 1607 | | FlowId, |
| | 4 | 1608 | | name, |
| | 4 | 1609 | | correlationId); |
| | | 1610 | | } |
| | | 1611 | | else |
| | | 1612 | | { |
| | 4 | 1613 | | _logger.LogWarning( |
| | 4 | 1614 | | "Flow {FlowId} lost its execution lease while step '{Step}' held a claimed response for correlationI |
| | 4 | 1615 | | FlowId, |
| | 4 | 1616 | | name, |
| | 4 | 1617 | | correlationId, |
| | 4 | 1618 | | found ? skipReason : "the ledger no longer exists"); |
| | | 1619 | | } |
| | 8 | 1620 | | } |
| | 0 | 1621 | | catch (Exception ex) |
| | | 1622 | | { |
| | | 1623 | | // The takeover signal is raised by the caller regardless; losing this write only means |
| | | 1624 | | // the step restarts as it did before. |
| | 0 | 1625 | | _logger.LogError( |
| | 0 | 1626 | | ex, |
| | 0 | 1627 | | "Flow {FlowId} could not checkpoint the claimed response for step '{Step}' after losing its execution le |
| | 0 | 1628 | | FlowId, |
| | 0 | 1629 | | name); |
| | 0 | 1630 | | } |
| | | 1631 | | |
| | 8 | 1632 | | if (!applied) |
| | 4 | 1633 | | return; |
| | | 1634 | | |
| | 4 | 1635 | | step.Completed = true; |
| | 4 | 1636 | | step.ResultJson = resultJson; |
| | 4 | 1637 | | step.PendingCorrelationId = null; |
| | 4 | 1638 | | step.PendingPayloadTypeFullName = null; |
| | 4 | 1639 | | step.CompletedAtUtc = completedAtUtc; |
| | 8 | 1640 | | } |
| | | 1641 | | |
| | | 1642 | | internal Task FlushProgressAsync() |
| | | 1643 | | { |
| | | 1644 | | // The body returned normally although a park failed — flow code swallowed the throw. The |
| | | 1645 | | // run is neither finished nor parked, so the executor must not mark it Succeeded: the |
| | | 1646 | | // failure surfaces here, where the executor awaits the body's outcome. |
| | 956 | 1647 | | if (_parkFailure is { } failure) |
| | 0 | 1648 | | return Task.FromException(failure.SourceException); |
| | | 1649 | | |
| | 956 | 1650 | | return _progressDirty ? SaveAsync(CancellationToken.None) : Task.CompletedTask; |
| | | 1651 | | } |
| | | 1652 | | |
| | | 1653 | | private async Task SaveAsync(CancellationToken cancellationToken, Exception? cause = null, TimeSpan? ttl = null) |
| | | 1654 | | { |
| | 8656 | 1655 | | _state.UpdatedAtUtc = UtcNow; |
| | 8656 | 1656 | | await _lease.SaveAsync(_state, ttl ?? _options.StateExpiry, cancellationToken, cause).ConfigureAwait(false); |
| | | 1657 | | |
| | 8636 | 1658 | | _progressDirty = false; |
| | 8636 | 1659 | | _lastPersistenceUtc = UtcNow; |
| | 8636 | 1660 | | WarnIfLedgerLarge(); |
| | 8636 | 1661 | | } |
| | | 1662 | | |
| | | 1663 | | /// <summary> |
| | | 1664 | | /// Every checkpoint rewrites the whole ledger, so a run whose steps retain sizeable results |
| | | 1665 | | /// pays a persistence cost that grows with each completed step (about N²/2 step-results |
| | | 1666 | | /// serialized over a run of N similar steps) until it hits the store's hard cap. The |
| | | 1667 | | /// <see cref="DurableFlowOptions.LedgerSizeWarningBytes"/> threshold turns that curve into an |
| | | 1668 | | /// early operator signal: one warning when it is first crossed, another at each doubling. |
| | | 1669 | | /// </summary> |
| | | 1670 | | private void WarnIfLedgerLarge() |
| | | 1671 | | { |
| | 8636 | 1672 | | if (_nextLedgerSizeWarningChars == long.MaxValue) |
| | 0 | 1673 | | return; |
| | | 1674 | | |
| | 8636 | 1675 | | var estimate = FlowStateJson.EstimateLedgerChars(_state); |
| | 8636 | 1676 | | if (estimate < _nextLedgerSizeWarningChars) |
| | 8632 | 1677 | | return; |
| | | 1678 | | |
| | 4 | 1679 | | _logger.LogWarning( |
| | 4 | 1680 | | "Durable flow {FlowId} ledger is roughly {LedgerBytes} bytes over {StepCount} step(s), past the {Threshold}- |
| | 4 | 1681 | | FlowId, |
| | 4 | 1682 | | estimate, |
| | 4 | 1683 | | _state.Steps?.Count ?? 0, |
| | 4 | 1684 | | _options.LedgerSizeWarningBytes); |
| | | 1685 | | |
| | | 1686 | | // Next warning at the next doubling of the CURRENT size (a single huge result may have |
| | | 1687 | | // skipped several thresholds at once), saturating instead of overflowing. |
| | 4 | 1688 | | _nextLedgerSizeWarningChars = estimate > long.MaxValue / 2 ? long.MaxValue - 1 : estimate * 2; |
| | 4 | 1689 | | } |
| | | 1690 | | |
| | | 1691 | | private async Task<TResponse> WaitForResponseAsync<TResponse>(Task<TResponse> responseTask, CancellationToken cancel |
| | | 1692 | | { |
| | 1503 | 1693 | | if (!cancellationToken.CanBeCanceled && !_hostStopping.CanBeCanceled) |
| | 1495 | 1694 | | return await responseTask.WaitAsync(_lease.LostToken).ConfigureAwait(false); |
| | | 1695 | | |
| | | 1696 | | // Host stop ends the park like the caller's token does (see WaitInProcessAsync): it lands |
| | | 1697 | | // in the awaited step's wait-side cancellation branch, which settles the handoff and keeps |
| | | 1698 | | // the breadcrumb, so the redelivered execution re-attaches to the same correlation id. |
| | 8 | 1699 | | using var linked = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, _lease.LostToken, _hostSto |
| | | 1700 | | try |
| | | 1701 | | { |
| | 8 | 1702 | | return await responseTask.WaitAsync(linked.Token).ConfigureAwait(false); |
| | | 1703 | | } |
| | 6 | 1704 | | catch (OperationCanceledException ex) when (_hostStopping.IsCancellationRequested |
| | 6 | 1705 | | && !cancellationToken.IsCancellationRequested |
| | 6 | 1706 | | && !_lease.LostToken.IsCancellationRequested) |
| | | 1707 | | { |
| | 2 | 1708 | | throw HostStopping(ex); |
| | | 1709 | | } |
| | 1456 | 1710 | | } |
| | | 1711 | | |
| | | 1712 | | private static TResult DeserializeResult<TResult>(string? resultJson) |
| | 306 | 1713 | | => resultJson is null ? default! : JsonSafety.SafeDeserialize<TResult>(resultJson)!; |
| | | 1714 | | } |