< Summary - AsyncResponse (Release / net8.0+net10.0 / unit+integration)

Information
Class: AsyncResponse.DurableFlowContext
Assembly: AsyncResponse.Core
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/AsyncResponse.Core/DurableFlowContext.cs
Line coverage
95%
Covered lines: 257
Uncovered lines: 12
Coverable lines: 269
Total lines: 575
Line coverage: 95.5%
Branch coverage
83%
Covered branches: 90
Total branches: 108
Branch coverage: 83.3%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.ctor(...)100%11100%
get_IsSuspended()100%11100%
get_FlowId()100%11100%
StepAsync()100%22100%
StepAsync()100%22100%
AwaitStepAsync<TResponse>(...)100%11100%
AwaitStepAsync<TResponse>(...)100%11100%
AwaitStepAsync<TResponse>(...)100%11100%
ReportProgressAsync(...)100%44100%
GetValue<TValue>(...)50%44100%
SetValueAsync<TValue>(...)100%22100%
AwaitChildFlowAsync()95.83%252489.09%
AwaitStepCoreAsync()73.33%303092.16%
CreateWaiterAsync()100%22100%
CreateChildState<TFlow, TInput>(...)100%11100%
EnqueueChildAsync(...)100%11100%
SuspendForChildAsync()100%11100%
ThrowIfSuspended()25%4475%
ThrowIfChildFailed(...)83.33%66100%
ThrowIfChildMismatched<TFlow, TInput>(...)92.86%1414100%
GetStep(...)100%44100%
CompleteStepAsync()75%4490%
FlushProgressAsync()100%22100%
SaveAsync()100%11100%
WaitForResponseAsync()100%22100%
DeserializeResult<TResult>(...)50%22100%

File(s)

/home/runner/work/AsyncResponse/AsyncResponse/src/AsyncResponse.Core/DurableFlowContext.cs

#LineLine coverage
 1using Microsoft.Extensions.Logging;
 2using System.Diagnostics.CodeAnalysis;
 3using System.Linq.Expressions;
 4
 5namespace AsyncResponse;
 6
 7/// <summary>
 8/// Runtime <see cref="IDurableFlowContext"/> bound to one execution of one flow run. Owns the
 9/// checkpointed-flow mechanics so flow code doesn't have to: step guards, result memoization, the
 10/// pending-correlation-id breadcrumb, fresh-start vs re-attach, and the durable resume/failure
 11/// callbacks that point back at the flow executor.
 12/// <para>
 13/// Not thread-safe by design: a flow body runs sequentially, and <c>until</c> predicates run on
 14/// the channel's dispatch path only while the flow itself is parked awaiting that same step.
 15/// </para>
 16/// </summary>
 17internal sealed class DurableFlowContext : IDurableFlowContext
 18{
 19    private readonly FlowState _state;
 20    private readonly IFlowStateStore _store;
 21    private readonly IAsyncResponseBuilder _builder;
 22    private readonly AsyncResponseContextPropagation _propagation;
 23    private readonly DurableFlowOptions _options;
 24    private readonly IAsyncResponseSubscriber _subscriber;
 25    private readonly IRecoverableAsyncResponseSubscriber? _recoverableSubscriber;
 26    private readonly ILogger _logger;
 27    private readonly FlowExecutionLease _lease;
 28    private bool _suspended;
 29    private bool _progressDirty;
 30    private DateTime _lastPersistenceUtc;
 31
 32    /// <summary>Creates the context for one execution of the given run.</summary>
 333    public DurableFlowContext(
 334        FlowState state,
 335        IFlowStateStore store,
 336        IAsyncResponseBuilder builder,
 337        AsyncResponseContextPropagation propagation,
 338        DurableFlowOptions options,
 339        IAsyncResponseSubscriber subscriber,
 340        IRecoverableAsyncResponseSubscriber? recoverableSubscriber,
 341        ILogger logger,
 342        FlowExecutionLease lease)
 43    {
 344        _state = state;
 345        _store = store;
 346        _builder = builder;
 347        _propagation = propagation;
 348        _options = options;
 349        _subscriber = subscriber;
 350        _recoverableSubscriber = recoverableSubscriber;
 351        _logger = logger;
 352        _lease = lease;
 353    }
 54
 355    internal bool IsSuspended => _suspended;
 56
 57    /// <inheritdoc />
 358    public string FlowId => _state.FlowId!;
 59
 60    /// <inheritdoc />
 61    public async Task StepAsync(string name, Func<Task> step, CancellationToken cancellationToken = default)
 62    {
 363        ThrowIfSuspended();
 364        ArgumentException.ThrowIfNullOrWhiteSpace(name);
 365        ArgumentNullException.ThrowIfNull(step);
 66
 367        var checkpoint = GetStep(name);
 368        if (checkpoint.Completed)
 269            return;
 70
 371        cancellationToken.ThrowIfCancellationRequested();
 372        await step().ConfigureAwait(false);
 373        _lease.ThrowIfLost();
 374        await CompleteStepAsync(name, checkpoint, resultJson: null, cancellationToken).ConfigureAwait(false);
 375    }
 76
 77    /// <inheritdoc />
 78    public async Task<TResult> StepAsync<TResult>(string name, Func<Task<TResult>> step, CancellationToken cancellationT
 79    {
 380        ThrowIfSuspended();
 381        ArgumentException.ThrowIfNullOrWhiteSpace(name);
 382        ArgumentNullException.ThrowIfNull(step);
 83
 384        var checkpoint = GetStep(name);
 385        if (checkpoint.Completed)
 286            return DeserializeResult<TResult>(checkpoint.ResultJson);
 87
 388        cancellationToken.ThrowIfCancellationRequested();
 389        var result = await step().ConfigureAwait(false);
 390        _lease.ThrowIfLost();
 391        await CompleteStepAsync(name, checkpoint, AsyncResponseJson.Serialize(result), cancellationToken).ConfigureAwait
 392        return result;
 393    }
 94
 95    /// <inheritdoc />
 96    public Task<TResponse> AwaitStepAsync<TResponse>(
 97        string name,
 98        Func<string, Task> trigger,
 99        TimeSpan? timeout = null,
 100        CancellationToken cancellationToken = default) where TResponse : IAsyncResponsePayload
 2101        => AwaitStepCoreAsync<TResponse>(name, trigger, until: null, timeout, cancellationToken);
 102
 103    /// <inheritdoc />
 104    public Task<TResponse> AwaitStepAsync<TResponse>(
 105        string name,
 106        Func<string, Task> trigger,
 107        Func<TResponse, bool> until,
 108        TimeSpan? timeout = null,
 109        CancellationToken cancellationToken = default) where TResponse : IAsyncResponsePayload
 110    {
 3111        ArgumentNullException.ThrowIfNull(until);
 3112        return AwaitStepCoreAsync<TResponse>(name, trigger, payload => new ValueTask<bool>(until(payload)), timeout, can
 113    }
 114
 115    /// <inheritdoc />
 116    public Task<TResponse> AwaitStepAsync<TResponse>(
 117        string name,
 118        Func<string, Task> trigger,
 119        Func<TResponse, Task<bool>> until,
 120        TimeSpan? timeout = null,
 121        CancellationToken cancellationToken = default) where TResponse : IAsyncResponsePayload
 122    {
 3123        ArgumentNullException.ThrowIfNull(until);
 3124        return AwaitStepCoreAsync<TResponse>(name, trigger, payload => new ValueTask<bool>(until(payload)), timeout, can
 125    }
 126
 127    /// <inheritdoc />
 128    public Task ReportProgressAsync(string message, CancellationToken cancellationToken = default)
 129    {
 3130        ThrowIfSuspended();
 3131        _state.LastMessage = message;
 3132        var now = DateTime.UtcNow;
 3133        if (_options.ProgressPersistenceInterval <= TimeSpan.Zero
 3134            || now - _lastPersistenceUtc >= _options.ProgressPersistenceInterval)
 2135            return SaveAsync(cancellationToken);
 136
 3137        _progressDirty = true;
 3138        return Task.CompletedTask;
 139    }
 140
 141    /// <inheritdoc />
 142    public TValue? GetValue<TValue>(string key)
 143    {
 2144        ThrowIfSuspended();
 2145        ArgumentException.ThrowIfNullOrWhiteSpace(key);
 2146        return _state.Values is not null && _state.Values.TryGetValue(key, out var json)
 2147            ? JsonSafety.SafeDeserialize<TValue>(json)
 2148            : default;
 149    }
 150
 151    /// <inheritdoc />
 152    public Task SetValueAsync<TValue>(string key, TValue value, CancellationToken cancellationToken = default)
 153    {
 3154        ThrowIfSuspended();
 3155        ArgumentException.ThrowIfNullOrWhiteSpace(key);
 3156        var values = _state.Values ??= new Dictionary<string, string>(StringComparer.Ordinal);
 3157        values[key] = AsyncResponseJson.Serialize(value);
 3158        return SaveAsync(cancellationToken);
 159    }
 160
 161    /// <inheritdoc />
 162    public async Task<FlowState> AwaitChildFlowAsync<[DynamicallyAccessedMembers(DynamicallyAccessedMemberTypes.PublicCo
 163        string name,
 164        TInput input,
 165        string? flowId = null,
 166        bool failOnChildFailure = true,
 167        CancellationToken cancellationToken = default)
 168        where TFlow : class, IDurableFlow<TInput>
 169    {
 2170        ThrowIfSuspended();
 2171        ArgumentException.ThrowIfNullOrWhiteSpace(name);
 2172        ArgumentNullException.ThrowIfNull(input);
 2173        if (flowId is not null)
 2174            ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 175
 2176        var checkpoint = GetStep(name);
 2177        var requestedChildFlowId = flowId ?? $"{FlowId}:{name}";
 2178        var breadcrumb = checkpoint.ChildFlowId;
 2179        if (breadcrumb is not null && !string.Equals(breadcrumb, requestedChildFlowId, StringComparison.Ordinal))
 180        {
 2181            throw new DurableFlowFailedException(
 2182                $"Step '{name}' of flow '{FlowId}' is already bound to child flow id '{breadcrumb}', " +
 2183                $"but this execution requested '{requestedChildFlowId}'. A durable step must keep the same child id on e
 184        }
 185
 2186        var childFlowId = breadcrumb ?? requestedChildFlowId;
 2187        var inputJson = AsyncResponseJson.Serialize(input);
 2188        if (checkpoint.Completed)
 189        {
 2190            var completedChild = DeserializeResult<FlowState>(checkpoint.ResultJson)
 2191                ?? throw new DurableFlowFailedException(
 2192                    $"Completed child step '{name}' of flow '{FlowId}' has no child-state snapshot.");
 2193            ThrowIfChildMismatched<TFlow, TInput>(completedChild, childFlowId, name, inputJson);
 0194            ThrowIfChildFailed(completedChild, failOnChildFailure);
 0195            return completedChild;
 196        }
 197
 2198        var child = await _store.LoadAsync(childFlowId, cancellationToken).ConfigureAwait(false);
 2199        if (child is null)
 200        {
 2201            if (breadcrumb is not null)
 202            {
 203                // The breadcrumb is persisted only after the child state exists, so a missing child
 204                // here means its ledger expired (StateExpiry) or was deleted while this parent was
 205                // suspended. Its outcome is unknowable; re-running it blind would re-execute side
 206                // effects of a possibly-completed run. Fail deterministically instead.
 2207                throw new DurableFlowFailedException(
 2208                    $"Child flow '{childFlowId}' has no state (expired or deleted) while parent flow '{FlowId}' was wait
 2209                    "Its outcome is unknown, so it is not re-run automatically. Size DurableFlowOptions.StateExpiry beyo
 2210                    "or start a new parent run to re-execute the work.");
 211            }
 212
 213            // Create the child BEFORE persisting the breadcrumb: "breadcrumb exists" must always
 214            // imply "child state existed", which keeps the expired-child check above sound. A crash
 215            // between the two writes is safe — the child id is deterministic, so the re-delivered
 216            // parent execution loads this child instead of re-creating it.
 2217            child = CreateChildState<TFlow, TInput>(childFlowId, name, inputJson);
 2218            if (await FlowStateConcurrency.TryCreateAsync(
 2219                    _store,
 2220                    childFlowId,
 2221                    child,
 2222                    _options.StateExpiry,
 2223                    cancellationToken).ConfigureAwait(false))
 224            {
 2225                _logger.LogDebug("Flow {FlowId} started child flow {ChildFlowId} for step '{Step}'.", FlowId, childFlowI
 226            }
 227            else
 228            {
 0229                child = await _store.LoadAsync(childFlowId, cancellationToken).ConfigureAwait(false)
 0230                    ?? throw new InvalidOperationException($"Child flow '{childFlowId}' was created concurrently but cou
 0231                ThrowIfChildMismatched<TFlow, TInput>(child, childFlowId, name, inputJson);
 232            }
 233        }
 234        else
 235        {
 2236            ThrowIfChildMismatched<TFlow, TInput>(child, childFlowId, name, inputJson);
 237        }
 238
 2239        if (breadcrumb is null)
 240        {
 2241            checkpoint.ChildFlowId = childFlowId;
 2242            checkpoint.Faulted = false;
 2243            checkpoint.Message = $"Waiting for child flow '{childFlowId}'.";
 2244            await SaveAsync(cancellationToken).ConfigureAwait(false);
 245        }
 246
 2247        switch (child.Status)
 248        {
 249            case FlowRunStatus.Succeeded:
 2250                await CompleteStepAsync(name, checkpoint, FlowStateJson.SerializeSnapshot(child), cancellationToken).Con
 2251                return child;
 252
 253            case FlowRunStatus.Failed:
 2254                checkpoint.Message = child.LastMessage;
 2255                await CompleteStepAsync(name, checkpoint, FlowStateJson.SerializeSnapshot(child), cancellationToken, fau
 2256                ThrowIfChildFailed(child, failOnChildFailure);
 2257                return child;
 258
 259            default:
 2260                await SuspendForChildAsync(childFlowId, cancellationToken).ConfigureAwait(false);
 0261                throw new InvalidOperationException("Unreachable.");
 262        }
 2263    }
 264
 265    private async Task<TResponse> AwaitStepCoreAsync<TResponse>(
 266        string name,
 267        Func<string, Task> trigger,
 268        Func<TResponse, ValueTask<bool>>? until,
 269        TimeSpan? timeout,
 270        CancellationToken cancellationToken) where TResponse : IAsyncResponsePayload
 271    {
 3272        ThrowIfSuspended();
 3273        ArgumentException.ThrowIfNullOrWhiteSpace(name);
 3274        ArgumentNullException.ThrowIfNull(trigger);
 275
 3276        var checkpoint = GetStep(name);
 3277        if (checkpoint.Completed)
 2278            return DeserializeResult<TResponse>(checkpoint.ResultJson);
 279
 280        // Re-attach when a previous execution already triggered this step and died waiting; start
 281        // fresh when there is no breadcrumb or the last attempt faulted (steps are idempotent).
 3282        var reattach = checkpoint.PendingCorrelationId is not null && !checkpoint.Faulted;
 3283        var correlationId = reattach
 3284            ? checkpoint.PendingCorrelationId!
 3285            : AsyncResponseContext.GenerateCorrelationId();
 3286        var stepTimeout = timeout ?? _options.DefaultStepTimeout;
 287
 3288        var waiter = await CreateWaiterAsync(correlationId, until, stepTimeout, name).ConfigureAwait(false);
 3289        var triggerCompleted = reattach;
 290        try
 291        {
 3292            if (!reattach)
 293            {
 294                // Persist the breadcrumb AFTER the registration exists and BEFORE the send:
 295                // "breadcrumb persisted" therefore implies "someone is listening", so a crash on
 296                // either side of the send re-attaches (or times out and restarts the idempotent
 297                // step) — never a lost run, never a double-send.
 3298                checkpoint.PendingCorrelationId = correlationId;
 3299                checkpoint.Faulted = false;
 3300                checkpoint.Message = null;
 3301                await SaveAsync(cancellationToken).ConfigureAwait(false);
 302
 3303                await trigger(correlationId).ConfigureAwait(false);
 3304                triggerCompleted = true;
 305            }
 2306            else if (_logger.IsEnabled(LogLevel.Debug))
 307            {
 0308                _logger.LogDebug(
 0309                    "Flow {FlowId} step '{Step}' re-attaching to in-flight correlationId {CorrelationId}.",
 0310                    FlowId, name, correlationId);
 311            }
 312
 3313            var response = await WaitForResponseAsync(waiter.ResponseTask, cancellationToken).ConfigureAwait(false);
 314
 3315            checkpoint.PendingCorrelationId = null;
 316            // Deliberately NOT the caller's token: once the response is claimed from the channel
 317            // it exists nowhere else, so the completion checkpoint must not be interruptible — a
 318            // cancellation here used to leave `pending` set with the response already consumed,
 319            // and the redelivered execution re-attached to a correlation id nothing could answer.
 3320            await CompleteStepAsync(name, checkpoint, AsyncResponseJson.Serialize(response), CancellationToken.None).Con
 3321            return response;
 322        }
 2323        catch (OperationCanceledException ex) when (triggerCompleted)
 324        {
 325            // A lost lease surfaces as cancellation of the linked wait; convert it with the wait
 326            // failure attached so the takeover signal does not discard the real cause.
 2327            _lease.ThrowIfLost(ex);
 328
 329            // SETTLE the handoff before deciding. A point-in-time IsCompletedSuccessfully check
 330            // raced the channel's dispatch: the response could win the task a moment after the
 331            // check, leaving a consumed response behind a still-pending ledger. Disposing the
 332            // waiter cancels its response task unless something already completed it (the channel
 333            // contract since the dispose-cancels fix), so after this await the task is TERMINAL
 334            // and the decision below is the race's single authoritative outcome. The finally's
 335            // second dispose is a no-op behind the subscription's cleanup latch.
 2336            await waiter.DisposeAsync().ConfigureAwait(false);
 337
 2338            if (waiter.ResponseTask.IsCompletedSuccessfully)
 339            {
 340                // Delivery won the settlement: the channel claimed and acked that message — it
 341                // exists nowhere else, and re-attaching to its consumed correlation id would park
 342                // the run until the step timeout. The checkpoint therefore wins over the
 343                // cancellation: persist the received payload and return it; the caller's token
 344                // gets its say again at the next step boundary.
 2345                var received = waiter.ResponseTask.Result;
 2346                checkpoint.PendingCorrelationId = null;
 2347                await CompleteStepAsync(name, checkpoint, AsyncResponseJson.Serialize(received), CancellationToken.None)
 2348                return received;
 349            }
 350
 2351            if (waiter.ResponseTask.IsFaulted)
 352            {
 353                // The wait FAULTED — a throwing Until predicate (possibly between the catch
 354                // filter and the settlement), or the disposal drain abandoning a wedged delivery
 355                // as AsyncResponseIndeterminateDeliveryException. Either way the message may be
 356                // consumed: restart the idempotent step fresh, exactly like the general fault
 357                // path below. The checkpoint records the fault's own message (not the
 358                // cancellation's) so the ledger says WHY the step restarts.
 2359                var fault = waiter.ResponseTask.Exception?.GetBaseException();
 2360                checkpoint.Faulted = true;
 2361                checkpoint.Message = fault?.Message ?? ex.Message;
 2362                await SaveAsync(CancellationToken.None, cause: fault ?? ex).ConfigureAwait(false);
 2363                throw;
 364            }
 365
 366            // Cancellation won the settlement (the task is now canceled; nothing was delivered).
 367            // WAIT-SIDE cancellation is infrastructure, not a step verdict: the channel cancels
 368            // in-flight waiters when it is disposed at host shutdown, and the caller's token
 369            // means "stop this execution", not "the step failed" — the remote operation is still
 370            // in flight. The persisted breadcrumb must survive untouched so the redelivered
 371            // execution RE-ATTACHES to the same correlation id; marking the checkpoint faulted
 372            // here turned every graceful shutdown mid-await into a fresh-correlation restart that
 373            // re-sent the remote request. (A response that never arrives still faults via the
 374            // step timeout.)
 375            //
 376            // The filter keeps this branch away from TRIGGER-thrown cancellation (an HttpClient
 377            // timeout surfaces as TaskCanceledException): the request may never have left the
 378            // process, so that case falls through to the fault path below and restarts fresh.
 2379            throw;
 0380        }
 2381        catch (Exception ex)
 382        {
 383            // Timeout, trigger failure (including trigger-thrown cancellation), or a faulted
 384            // wait: record it so the next execution restarts this step fresh instead of
 385            // re-attaching to a dead correlation id. The original failure rides along as `cause`
 386            // so a rejected save cannot displace it.
 2387            checkpoint.Faulted = true;
 2388            checkpoint.Message = ex.Message;
 2389            await SaveAsync(CancellationToken.None, cause: ex).ConfigureAwait(false);
 2390            throw;
 391        }
 392        finally
 393        {
 3394            await waiter.DisposeAsync().ConfigureAwait(false);
 395        }
 3396    }
 397
 398    private async Task<IAsyncResponseWaiter<TResponse>> CreateWaiterAsync<TResponse>(
 399        string correlationId,
 400        Func<TResponse, ValueTask<bool>>? until,
 401        TimeSpan? timeout,
 402        string stepName) where TResponse : IAsyncResponsePayload
 403    {
 3404        if (_recoverableSubscriber is not null)
 405        {
 406            // The durable safety net: a response landing while no process is executing this flow
 407            // checkpoints the terminal payload and re-enqueues the run, or terminally fails it —
 408            // the same at-least-once, idempotency-required contract as hand-registered callbacks.
 3409            var flowId = FlowId;
 3410            Expression<Func<IDurableFlowExecutor, Task>> resume = executor => executor.RecoverAsync(
 3411                flowId,
 3412                Placeholder.Payload<TResponse>()!,
 3413                Placeholder.CorrelationId());
 3414            Expression<Func<IDurableFlowExecutor, Task>> failure = executor => executor.FailAsync(flowId, Placeholder.Ex
 415
 3416            return await _recoverableSubscriber.CreateRecoverableResponseWaiter(
 3417                correlationId,
 3418                CallbackExpressionConverter.ToReflectionCall(resume),
 3419                CallbackExpressionConverter.ToReflectionCall(failure),
 3420                until,
 3421                timeout).ConfigureAwait(false);
 422        }
 423
 2424        _logger.LogDebug(
 2425            "Flow {FlowId} step '{Step}': the configured channel exposes no recoverable subscriber; lost-subscriber reco
 2426            FlowId, stepName);
 427
 2428        return await _subscriber.CreateResponseWaiter(correlationId, until, timeout).ConfigureAwait(false);
 3429    }
 430
 431    private FlowState CreateChildState<TFlow, TInput>(string flowId, string parentStepName, string inputJson)
 432    {
 2433        var now = DateTime.UtcNow;
 2434        return new FlowState
 2435        {
 2436            FlowId = flowId,
 2437            FlowTypeName = typeof(TFlow).FullName,
 2438            InputTypeName = typeof(TInput).FullName,
 2439            InputJson = inputJson,
 2440            Status = FlowRunStatus.Running,
 2441            LastMessage = $"Child flow started by {FlowId}.",
 2442            CreatedAtUtc = now,
 2443            UpdatedAtUtc = now,
 2444            ParentFlowId = FlowId,
 2445            ParentStepName = parentStepName,
 2446            Context = _propagation.Capture()
 2447        };
 448    }
 449
 450    private Task EnqueueChildAsync(string childFlowId)
 451    {
 2452        var id = childFlowId;
 2453        return _builder.EnqueueWorkerAsync<IDurableFlowExecutor>(executor => executor.ExecuteAsync(id));
 454    }
 455
 456    private async Task SuspendForChildAsync(string childFlowId, CancellationToken cancellationToken)
 457    {
 458        // Persist the suspension BEFORE the child becomes runnable: once the child is enqueued it
 459        // can complete and re-execute this parent on another worker at any moment, and a save after
 460        // that point would clobber the re-execution's newer checkpoints with this stale snapshot.
 461        // The executor therefore does NOT save again on the suspension path.
 2462        _suspended = true;
 2463        _state.LastMessage = $"Flow {FlowId} suspended waiting for child flow {childFlowId}.";
 2464        await SaveAsync(cancellationToken).ConfigureAwait(false);
 2465        await EnqueueChildAsync(childFlowId).ConfigureAwait(false);
 2466        throw new DurableFlowSuspendedException(_state.LastMessage);
 467    }
 468
 469    private void ThrowIfSuspended()
 470    {
 3471        _lease.ThrowIfLost();
 3472        if (_suspended)
 0473            throw new DurableFlowSuspendedException(_state.LastMessage ?? $"Flow {FlowId} is suspended.");
 3474    }
 475
 476    private static void ThrowIfChildFailed(FlowState child, bool failOnChildFailure)
 477    {
 2478        if (failOnChildFailure && child.Status == FlowRunStatus.Failed)
 2479            throw new DurableFlowFailedException($"Child flow '{child.FlowId}' failed: {child.LastMessage ?? "no message
 2480    }
 481
 482    private void ThrowIfChildMismatched<TFlow, TInput>(
 483        FlowState child,
 484        string childFlowId,
 485        string stepName,
 486        string requestedInputJson)
 487    {
 488        // A child id is owned by exactly one parent: the notification that resumes a suspended
 489        // parent follows the child's single ParentFlowId, so a second parent awaiting the same id
 490        // would suspend and never wake. Reject collisions loudly instead of parking forever.
 2491        if (!string.Equals(child.ParentFlowId, FlowId, StringComparison.Ordinal))
 492        {
 2493            var owner = child.ParentFlowId is null ? "a run not started by AwaitChildFlowAsync" : $"parent flow '{child.
 2494            throw new DurableFlowFailedException(
 2495                $"Step '{stepName}' of flow '{FlowId}' awaits child flow id '{childFlowId}', but that id belongs to {own
 2496                "Child flow ids are exclusive to the parent that started them — pass a flowId that is unique per parent 
 2497                "(the default '{parentFlowId}:{stepName}' id is always safe).");
 498        }
 499
 2500        if (!string.Equals(child.FlowId, childFlowId, StringComparison.Ordinal)
 2501            || !string.Equals(child.ParentStepName, stepName, StringComparison.Ordinal))
 502        {
 2503            throw new DurableFlowFailedException(
 2504                $"Child flow id '{childFlowId}' is bound to a different child step than '{stepName}' of parent flow '{Fl
 2505                "A child id is exclusive to one parent step.");
 506        }
 507
 2508        if (!string.Equals(child.FlowTypeName, typeof(TFlow).FullName, StringComparison.Ordinal))
 509        {
 2510            throw new DurableFlowFailedException(
 2511                $"Step '{stepName}' of flow '{FlowId}' awaits child flow id '{childFlowId}' as {typeof(TFlow).FullName},
 2512                $"but the persisted run is {child.FlowTypeName}. The flowId collides with a different flow — use a uniqu
 513        }
 514
 2515        if (!string.Equals(child.InputTypeName, typeof(TInput).FullName, StringComparison.Ordinal)
 2516            || !FlowStateJson.JsonEquivalent(child.InputJson, requestedInputJson))
 517        {
 2518            throw new DurableFlowFailedException(
 2519                $"Step '{stepName}' of flow '{FlowId}' requested child flow id '{childFlowId}' with a different input " 
 2520                "type or value than the persisted child. Replays must use semantically identical child input.");
 521        }
 2522    }
 523
 524    private FlowStepState GetStep(string name)
 525    {
 3526        var steps = _state.Steps ??= new Dictionary<string, FlowStepState>(StringComparer.Ordinal);
 3527        if (!steps.TryGetValue(name, out var step))
 528        {
 3529            step = new FlowStepState();
 3530            steps[name] = step;
 531        }
 532
 3533        return step;
 534    }
 535
 536    private async Task CompleteStepAsync(string name, FlowStepState step, string? resultJson, CancellationToken cancella
 537    {
 3538        step.Completed = true;
 3539        step.ResultJson = resultJson;
 3540        step.PendingCorrelationId = null;
 541        // A memoized failed child keeps Faulted = true so operators can spot the failure on the
 542        // step itself instead of digging through ResultJson.
 3543        step.Faulted = faulted;
 3544        step.CompletedAtUtc = DateTime.UtcNow;
 3545        _state.LastMessage = faulted ? $"Step '{name}' completed (child flow failed)." : $"Step '{name}' completed.";
 3546        await SaveAsync(cancellationToken).ConfigureAwait(false);
 547
 3548        if (_logger.IsEnabled(LogLevel.Debug))
 0549            _logger.LogDebug("Flow {FlowId} step '{Step}' completed.", FlowId, name);
 3550    }
 551
 552    internal Task FlushProgressAsync()
 3553        => _progressDirty ? SaveAsync(CancellationToken.None) : Task.CompletedTask;
 554
 555    private async Task SaveAsync(CancellationToken cancellationToken, Exception? cause = null)
 556    {
 3557        _state.UpdatedAtUtc = DateTime.UtcNow;
 3558        await _lease.SaveAsync(_state, _options.StateExpiry, cancellationToken, cause).ConfigureAwait(false);
 559
 3560        _progressDirty = false;
 3561        _lastPersistenceUtc = DateTime.UtcNow;
 3562    }
 563
 564    private async Task<TResponse> WaitForResponseAsync<TResponse>(Task<TResponse> responseTask, CancellationToken cancel
 565    {
 3566        if (!cancellationToken.CanBeCanceled)
 3567            return await responseTask.WaitAsync(_lease.LostToken).ConfigureAwait(false);
 568
 2569        using var linked = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, _lease.LostToken);
 2570        return await responseTask.WaitAsync(linked.Token).ConfigureAwait(false);
 3571    }
 572
 573    private static TResult DeserializeResult<TResult>(string? resultJson)
 2574        => resultJson is null ? default! : JsonSafety.SafeDeserialize<TResult>(resultJson)!;
 575}
Generated by: ReportGenerator 5.5.10.0
08/03/2026 - 08:59:44
GitHub | reportgenerator.io

Methods/Properties

.ctor(AsyncResponse.FlowState, AsyncResponse.IFlowStateStore, AsyncResponse.IAsyncResponseBuilder, AsyncResponse.AsyncResponseContextPropagation, AsyncResponse.DurableFlowOptions, AsyncResponse.IAsyncResponseSubscriber, AsyncResponse.IRecoverableAsyncResponseSubscriber, Microsoft.Extensions.Logging.ILogger, AsyncResponse.FlowExecutionLease)
get_IsSuspended()
get_FlowId()
StepAsync()
StepAsync()
AwaitStepAsync<TResponse>(string, System.Func<string, System.Threading.Tasks.Task>, System.Nullable<System.TimeSpan>, System.Threading.CancellationToken)
AwaitStepAsync<TResponse>(string, System.Func<string, System.Threading.Tasks.Task>, System.Func<TResponse, bool>, System.Nullable<System.TimeSpan>, System.Threading.CancellationToken)
AwaitStepAsync<TResponse>(string, System.Func<string, System.Threading.Tasks.Task>, System.Func<TResponse, System.Threading.Tasks.Task<bool>>, System.Nullable<System.TimeSpan>, System.Threading.CancellationToken)
ReportProgressAsync(string, System.Threading.CancellationToken)
GetValue<TValue>(string)
SetValueAsync<TValue>(string, TValue, System.Threading.CancellationToken)
AwaitChildFlowAsync()
AwaitStepCoreAsync()
CreateWaiterAsync()
CreateChildState<TFlow, TInput>(string, string, string)
EnqueueChildAsync(string)
SuspendForChildAsync()
ThrowIfSuspended()
ThrowIfChildFailed(AsyncResponse.FlowState, bool)
ThrowIfChildMismatched<TFlow, TInput>(AsyncResponse.FlowState, string, string, string)
GetStep(string)
CompleteStepAsync()
FlushProgressAsync()
SaveAsync()
WaitForResponseAsync()
DeserializeResult<TResult>(string)