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

Information
Class: AsyncResponse.DurableFlowService
Assembly: AsyncResponse.Core
File(s): /_/src/AsyncResponse.Core/DurableFlows.cs
Line coverage
93%
Covered lines: 121
Uncovered lines: 9
Coverable lines: 130
Total lines: 244
Line coverage: 93%
Branch coverage
94%
Covered branches: 17
Total branches: 18
Branch coverage: 94.4%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.ctor(...)100%22100%
StartAsync()83.33%6683.63%
PublishStartAsync()100%11100%
<PublishStartAsync()100%11100%
ResumeAsync()100%44100%
GetStateAsync()100%11100%
EnsureIdempotentStart(...)100%22100%

File(s)

/_/src/AsyncResponse.Core/DurableFlows.cs

#LineLine coverage
 1using Microsoft.Extensions.DependencyInjection;
 2using Microsoft.Extensions.Logging;
 3using System.Diagnostics.CodeAnalysis;
 4
 5namespace AsyncResponse;
 6
 7/// <inheritdoc cref="IDurableFlows" />
 8internal sealed class DurableFlowService : IDurableFlows
 9{
 10    private readonly IServiceScopeFactory _scopeFactory;
 11    private readonly IAsyncResponseBuilder _builder;
 12    private readonly AsyncResponseContextPropagation _propagation;
 13    private readonly DurableFlowOptions _options;
 14    private readonly ILogger<DurableFlowService> _logger;
 15    private readonly TimeProvider _timeProvider;
 16
 17    /// <summary>Creates the durable-flows starter.</summary>
 159718    public DurableFlowService(
 159719        IServiceScopeFactory scopeFactory,
 159720        IAsyncResponseBuilder builder,
 159721        AsyncResponseContextPropagation propagation,
 159722        DurableFlowOptions options,
 159723        ILogger<DurableFlowService> logger,
 159724        TimeProvider? timeProvider = null)
 25    {
 159726        _scopeFactory = scopeFactory;
 159727        _builder = builder;
 159728        _propagation = propagation;
 159729        _options = options;
 159730        FlowStateConcurrency.ValidateOptions(_options);
 159731        _options.ValidateInProcessPark();
 159132        _logger = logger;
 159133        _timeProvider = timeProvider ?? TimeProvider.System;
 159134    }
 35
 36    /// <inheritdoc />
 37    public async Task<string> StartAsync<[DynamicallyAccessedMembers(DynamicallyAccessedMemberTypes.PublicConstructors |
 38        TInput input,
 39        string? flowId = null,
 40        CancellationToken cancellationToken = default)
 41        where TFlow : class, IDurableFlow<TInput>
 42    {
 161043        ArgumentNullException.ThrowIfNull(input);
 161044        cancellationToken.ThrowIfCancellationRequested();
 160845        if (flowId is null)
 147046            flowId = $"flow-{AsyncResponseContext.GenerateCorrelationId()}";
 47        else
 13848            ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 49
 50        // Every id is validated BEFORE anything is published: the publish below is the start's
 51        // commit point, and a job for an id every store would reject must never leave the process.
 160452        FlowStateConcurrency.EnsurePortableFlowId(flowId);
 53
 158254        await using var scope = _scopeFactory.CreateAsyncScope();
 158255        var store = scope.ServiceProvider.GetRequiredService<IFlowStateStore>();
 56
 158257        var now = _timeProvider.GetUtcNow().UtcDateTime;
 158258        var inputJson = AsyncResponseJson.Serialize(input);
 158259        var state = new FlowState
 158260        {
 158261            FlowId = flowId,
 158262            FlowTypeName = typeof(TFlow).FullName,
 158263            InputTypeName = typeof(TInput).FullName,
 158264            InputJson = inputJson,
 158265            Status = FlowRunStatus.Running,
 158266            LastMessage = "Flow started.",
 158267            CreatedAtUtc = now,
 158268            UpdatedAtUtc = now,
 158269            Revision = 0,
 158270            Context = _propagation.Capture()
 158271        };
 72
 73        // PUBLISH FIRST, then create. The worker job carries the whole initial ledger, and
 74        // IDurableFlowExecutor.CreateAndExecuteAsync creates the ledger itself (insert-if-absent)
 75        // before executing — so the publish is the single durable commit point of a start:
 76        //  - a crash before the publish leaves nothing behind (the caller sees a fault and retries);
 77        //  - a crash after the publish leaves a job whose execution creates and runs the flow.
 78        // The previous order (create, then publish) had an unrecoverable gap: a process dying
 79        // between the two left a committed Running ledger with Attempts = 0 that nothing would
 80        // ever execute, and IFlowStateStore has no enumeration for a reconciler to go find it.
 81        // The publish still runs the retry ladder the ingress uses, and a publish that fails for
 82        // good surfaces the id (DurableFlowNotDispatchedException) — now with nothing persisted.
 158283        var id = flowId;
 158284        var initialStateJson = FlowStateJson.Serialize(state);
 158285        store.ValidateCreate(flowId, state, _options.StateExpiry);
 158086        await PublishStartAsync(
 158087            executor => executor.CreateAndExecuteAsync(id, initialStateJson),
 158088            id,
 158089            cancellationToken).ConfigureAwait(false);
 90
 91        // Normally the starter's own create makes state immediately queryable and reports an
 92        // explicit-id conflict to this caller. Losing the race to the executor or an identical
 93        // start is expected. After a transient store fault the published job creates the ledger;
 94        // a query can return null until it does. Deterministic size/argument rejection still
 95        // propagates, including from custom stores whose preflight uses the no-op default.
 96        bool created;
 97        try
 98        {
 157099            created = await FlowStateConcurrency.TryCreateAsync(
 1570100                store,
 1570101                flowId,
 1570102                state,
 1570103                _options.StateExpiry,
 1570104                cancellationToken).ConfigureAwait(false);
 1570105        }
 0106        catch (Exception ex) when (ex is not (FlowStateTooLargeException or ArgumentException)
 0107            && (ex is not OperationCanceledException || !cancellationToken.IsCancellationRequested))
 108        {
 0109            _logger.LogWarning(
 0110                ex,
 0111                "Durable flow {FlowId} start job is published but the starter could not write the ledger; the executor c
 0112                flowId);
 0113            return flowId;
 114        }
 115
 1570116        if (created)
 117        {
 890118            _logger.LogInformation("Started durable flow {FlowId} ({FlowType}).", flowId, typeof(TFlow).Name);
 890119            return flowId;
 120        }
 121
 680122        var existing = await store.LoadAsync(flowId, cancellationToken).ConfigureAwait(false);
 680123        if (existing is null)
 124        {
 125            // Lost the create to a ledger that has since expired: the published job's create wins
 126            // the next time round. Nothing for the caller to do.
 0127            _logger.LogWarning("Durable flow {FlowId} start job is published; the existing ledger is expired and the exe
 0128            return flowId;
 129        }
 130
 131        // Throws DurableFlowIdConflictException for different work; the executor drops the
 132        // already-published job on the same test.
 680133        EnsureIdempotentStart<TFlow, TInput>(existing, inputJson, flowId);
 134
 135        // A semantically identical retry: the published job re-enqueues the existing run
 136        // (completed steps skip) instead of creating a duplicate.
 664137        _logger.LogInformation("Durable flow {FlowId} already exists; the start job re-enqueues the existing run instead
 664138        return flowId;
 1554139    }
 140
 141    /// <summary>
 142    /// Publishes a start job through the ingress's retry ladder. A publish that still fails
 143    /// surfaces as <see cref="DurableFlowNotDispatchedException"/> carrying the id: nothing was
 144    /// persisted, so the caller simply retries the start (idempotent with the same id).
 145    /// </summary>
 146    private async Task PublishStartAsync(
 147        System.Linq.Expressions.Expression<Func<IDurableFlowExecutor, Task>> job,
 148        string flowId,
 149        CancellationToken cancellationToken)
 150    {
 151        try
 152        {
 1580153            await AsyncResponseRetry.ExecuteAsync(
 1580154                async token =>
 1580155                {
 1602156                    await _builder.EnqueueWorkerAsync(job, token).ConfigureAwait(false);
 1570157                    return true;
 1570158                },
 1580159                // Only the CALLER's cancellation ends the ladder. An OperationCanceledException
 1580160                // whose token is not the caller's is a transport or SDK timeout — brokers surface
 1580161                // those as TaskCanceledException all the time — and that is exactly the transient
 1580162                // shape this retry exists for. Excluding the whole exception type meant the most
 1580163                // common recoverable publish failure got zero retries. An envelope over the
 1580164                // ingress's size budget is deterministic (the same input serializes to the same
 1580165                // length): no attempt can succeed, so it is not retried either.
 26166                isTransient: ex => ex is not WorkerJobTooLargeException
 26167                    && (ex is not OperationCanceledException || !cancellationToken.IsCancellationRequested),
 1580168                maxAttempts: 4,
 1580169                baseDelay: TimeSpan.FromMilliseconds(250),
 1580170                maxDelay: TimeSpan.FromSeconds(2),
 1580171                cancellationToken,
 1580172                _timeProvider).ConfigureAwait(false);
 1570173        }
 4174        catch (WorkerJobTooLargeException ex)
 175        {
 176            // Not a dispatch failure to retry: the start job carries the initial ledger, and this
 177            // input serializes past what the consuming ingress accepts — it would be acknowledged
 178            // there without ever executing. Surfaced as itself (nothing was persisted) so the
 179            // caller can shrink the input or move it behind a claim check.
 4180            _logger.LogError(
 4181                ex,
 4182                "Durable flow {FlowId} could not be started: its start job ({SerializedLength} UTF-16 code units) exceed
 4183                flowId,
 4184                ex.SerializedLength,
 4185                ex.Limit);
 4186            throw;
 187        }
 6188        catch (Exception ex)
 189        {
 6190            _logger.LogError(
 6191                ex,
 6192                "Durable flow {FlowId} could not be started: its worker job was not published after retries. Nothing was
 6193                flowId);
 6194            throw new DurableFlowNotDispatchedException(flowId, ex);
 195        }
 1570196    }
 197
 198    /// <inheritdoc />
 199    public async Task ResumeAsync(string flowId, CancellationToken cancellationToken = default)
 200    {
 684201        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 202
 682203        await using var scope = _scopeFactory.CreateAsyncScope();
 682204        var store = scope.ServiceProvider.GetRequiredService<IFlowStateStore>();
 205
 682206        var state = await store.LoadAsync(flowId, cancellationToken).ConfigureAwait(false)
 682207            ?? throw new InvalidOperationException($"No flow state found for '{flowId}' (unknown, expired, or unreadable
 208
 680209        if (state.Status != FlowRunStatus.Running)
 210        {
 666211            _logger.LogDebug("Durable flow {FlowId} is already {Status}; ignoring resume.", flowId, state.Status);
 666212            return;
 213        }
 214
 14215        var id = flowId;
 14216        await _builder.EnqueueWorkerAsync<IDurableFlowExecutor>(
 14217            executor => executor.ExecuteAsync(id),
 14218            cancellationToken).ConfigureAwait(false);
 680219    }
 220
 221    /// <inheritdoc />
 222    public async Task<FlowState?> GetStateAsync(string flowId, CancellationToken cancellationToken = default)
 223    {
 4428224        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 225
 4428226        await using var scope = _scopeFactory.CreateAsyncScope();
 4428227        var store = scope.ServiceProvider.GetRequiredService<IFlowStateStore>();
 4428228        return await store.LoadAsync(flowId, cancellationToken).ConfigureAwait(false);
 4427229    }
 230
 231    private static void EnsureIdempotentStart<TFlow, TInput>(
 232        FlowState existing,
 233        string requestedInputJson,
 234        string flowId)
 235    {
 680236        if (FlowStateConcurrency.IsSameStart(existing, typeof(TFlow).FullName, typeof(TInput).FullName, requestedInputJs
 664237            return;
 238
 16239        throw new DurableFlowIdConflictException(
 16240            $"Durable flow id '{flowId}' is already bound to a different flow type or input. " +
 16241            "Idempotent retries must use the same TFlow, TInput, and semantically identical input value.");
 242    }
 243
 244}
Generated by: ReportGenerator 5.5.11.0
09/22/2026 - 09:41:04
GitHub | reportgenerator.io

Methods/Properties

.ctor(Microsoft.Extensions.DependencyInjection.IServiceScopeFactory,AsyncResponse.IAsyncResponseBuilder,AsyncResponse.AsyncResponseContextPropagation,AsyncResponse.DurableFlowOptions,Microsoft.Extensions.Logging.ILogger`1<AsyncResponse.DurableFlowService>,System.TimeProvider)
StartAsync()
PublishStartAsync()
<PublishStartAsync()
ResumeAsync()
GetStateAsync()
EnsureIdempotentStart(AsyncResponse.FlowState,System.String,System.String)