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

Information
Class: AsyncResponse.FlowExecutionLease
Assembly: AsyncResponse.Core
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/AsyncResponse.Core/FlowStateConcurrency.cs
Line coverage
96%
Covered lines: 90
Uncovered lines: 3
Coverable lines: 93
Total lines: 272
Line coverage: 96.7%
Branch coverage
85%
Covered branches: 12
Total branches: 14
Branch coverage: 85.7%
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_LostToken()100%11100%
ThrowIfLost(...)100%22100%
SaveAsync()50%2295.65%
CreateSaveRejectedExceptionAsync()100%4484.62%
RenewLoopAsync()75%44100%
MarkLost()100%11100%
DisposeAsync()100%22100%

File(s)

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

#LineLine coverage
 1using Microsoft.Extensions.Logging;
 2
 3namespace AsyncResponse;
 4
 5/// <summary>Coordinates atomic flow creation, optimistic updates, and one active executor per flow id.</summary>
 6internal static class FlowStateConcurrency
 7{
 8    private const int MaxUpdateAttempts = 8;
 9
 10    public static Task<bool> TryCreateAsync(
 11        IFlowStateStore store,
 12        string flowId,
 13        FlowState state,
 14        TimeSpan ttl,
 15        CancellationToken cancellationToken = default)
 16    {
 17        state.Revision = 0;
 18        return store.TryCreateAsync(flowId, state, ttl, cancellationToken);
 19    }
 20
 21    public static async Task<FlowExecutionLease?> TryAcquireExecutionLeaseAsync(
 22        IFlowStateStore store,
 23        string flowId,
 24        DurableFlowOptions options,
 25        ILogger logger,
 26        CancellationToken cancellationToken = default)
 27    {
 28        ValidateOptions(options);
 29
 30        var leaseId = Guid.NewGuid().ToString("N");
 31        if (!await store.TryAcquireLeaseAsync(
 32                flowId,
 33                leaseId,
 34                options.ExecutionLeaseDuration,
 35                cancellationToken).ConfigureAwait(false))
 36            return null;
 37
 38        return new FlowExecutionLease(store, flowId, leaseId, options, logger);
 39    }
 40
 41    public static async Task<bool> MutateAsync(
 42        IFlowStateStore store,
 43        string flowId,
 44        TimeSpan ttl,
 45        Func<FlowState, bool> mutate,
 46        CancellationToken cancellationToken = default)
 47    {
 48        for (var attempt = 0; attempt < MaxUpdateAttempts; attempt++)
 49        {
 50            var state = await store.LoadAsync(flowId, cancellationToken).ConfigureAwait(false);
 51            if (state is null)
 52                return false;
 53
 54            if (!mutate(state))
 55                return true;
 56
 57            var expectedRevision = state.Revision;
 58            state.Revision = checked(expectedRevision + 1);
 59            state.UpdatedAtUtc = DateTime.UtcNow;
 60            if (await store.TryUpdateAsync(
 61                    flowId,
 62                    state,
 63                    expectedRevision,
 64                    ttl,
 65                    leaseId: null,
 66                    cancellationToken).ConfigureAwait(false))
 67                return true;
 68        }
 69
 70        throw new InvalidOperationException(
 71            $"Durable flow '{flowId}' changed repeatedly while applying a recovery update; retry the operation.");
 72    }
 73
 74    internal static void ValidateOptions(DurableFlowOptions options)
 75    {
 76        if (options.StateExpiry <= TimeSpan.Zero)
 77            throw new InvalidOperationException($"{nameof(DurableFlowOptions)}.{nameof(options.StateExpiry)} must be pos
 78        if (options.DefaultStepTimeout is { } defaultStepTimeout && defaultStepTimeout <= TimeSpan.Zero)
 79            throw new InvalidOperationException($"{nameof(DurableFlowOptions)}.{nameof(options.DefaultStepTimeout)} must
 80        if (options.ExecutionLeaseDuration <= TimeSpan.Zero)
 81            throw new InvalidOperationException($"{nameof(DurableFlowOptions)}.{nameof(options.ExecutionLeaseDuration)} 
 82        if (options.ExecutionLeaseRenewInterval <= TimeSpan.Zero
 83            || options.ExecutionLeaseRenewInterval >= options.ExecutionLeaseDuration)
 84        {
 85            throw new InvalidOperationException(
 86                $"{nameof(DurableFlowOptions)}.{nameof(options.ExecutionLeaseRenewInterval)} must be positive and shorte
 87                $"{nameof(DurableFlowOptions.ExecutionLeaseDuration)}.");
 88        }
 89        if (options.ProgressPersistenceInterval < TimeSpan.Zero)
 90            throw new InvalidOperationException($"{nameof(DurableFlowOptions)}.{nameof(options.ProgressPersistenceInterv
 91    }
 92}
 93
 94/// <summary>One distributed durable-flow execution lease.</summary>
 95internal sealed class FlowExecutionLease : IAsyncDisposable
 96{
 97    private readonly IFlowStateStore _store;
 98    private readonly string _flowId;
 99    private readonly string _leaseId;
 100    private readonly DurableFlowOptions _options;
 101    private readonly ILogger _logger;
 3102    private readonly CancellationTokenSource _stop = new();
 3103    private readonly CancellationTokenSource _lost = new();
 104    private readonly Task _renewal;
 105    private DateTime _validUntilUtc;
 106    private int _disposed;
 107
 3108    public FlowExecutionLease(
 3109        IFlowStateStore store,
 3110        string flowId,
 3111        string leaseId,
 3112        DurableFlowOptions options,
 3113        ILogger logger)
 114    {
 3115        _store = store;
 3116        _flowId = flowId;
 3117        _leaseId = leaseId;
 3118        _options = options;
 3119        _logger = logger;
 3120        _validUntilUtc = DateTime.UtcNow.Add(options.ExecutionLeaseDuration);
 3121        _renewal = RenewLoopAsync();
 3122    }
 123
 3124    public CancellationToken LostToken => _lost.Token;
 125
 126    /// <summary>
 127    /// Throws when the lease is lost. <paramref name="cause"/> (e.g. the exception that made the
 128    /// caller check) is attached as the inner exception so the real failure is not discarded.
 129    /// </summary>
 130    public void ThrowIfLost(Exception? cause = null)
 131    {
 3132        if (_lost.IsCancellationRequested)
 2133            throw new InvalidOperationException($"Durable flow '{_flowId}' lost its execution lease; the worker will ret
 3134    }
 135
 136    public async Task SaveAsync(FlowState state, TimeSpan ttl, CancellationToken cancellationToken = default, Exception?
 137    {
 3138        ThrowIfLost(cause);
 3139        var expectedRevision = state.Revision;
 3140        state.Revision = checked(expectedRevision + 1);
 3141        state.UpdatedAtUtc = DateTime.UtcNow;
 142
 143        try
 144        {
 3145            if (await _store.TryUpdateAsync(
 3146                    _flowId,
 3147                    state,
 3148                    expectedRevision,
 3149                    ttl,
 3150                    _leaseId,
 3151                    cancellationToken).ConfigureAwait(false))
 3152                return;
 2153        }
 2154        catch
 155        {
 2156            state.Revision = expectedRevision;
 2157            MarkLost();
 158
 159            // The store exception propagates; keep the failure this save was recording from
 160            // vanishing with it.
 2161            if (cause is not null)
 0162                _logger.LogWarning(cause, "Durable flow '{FlowId}' failed to checkpoint; the failure it was recording is
 2163            throw;
 164        }
 165
 2166        state.Revision = expectedRevision;
 2167        MarkLost();
 2168        throw await CreateSaveRejectedExceptionAsync(expectedRevision, cause, cancellationToken).ConfigureAwait(false);
 3169    }
 170
 171    /// <summary>
 172    /// Builds the exception for a rejected checkpoint write. The store's compare-and-swap only
 173    /// returns <c>false</c>, so the reason is diagnosed with a best-effort re-read: a revision
 174    /// conflict — a concurrent lease-bypassing writer such as <c>RecoverAsync</c>, <c>FailAsync</c>,
 175    /// or an operator parking the run — is reported as such instead of as a lost lease, which sent
 176    /// operators hunting phantom lease problems. Behavior is unchanged either way: the lease is
 177    /// abandoned (<see cref="MarkLost"/> already ran) and the delivery retries from the last
 178    /// checkpoint; <paramref name="cause"/> rides along as the inner exception so the failure that
 179    /// triggered the save is not discarded.
 180    /// </summary>
 181    private async Task<InvalidOperationException> CreateSaveRejectedExceptionAsync(
 182        long expectedRevision,
 183        Exception? cause,
 184        CancellationToken cancellationToken)
 185    {
 2186        var reason = "its execution lease was no longer held (expired or taken over)";
 187        try
 188        {
 2189            var current = await _store.LoadAsync(_flowId, cancellationToken).ConfigureAwait(false);
 2190            if (current is null)
 2191                reason = "its ledger entry is gone (expired or deleted)";
 2192            else if (current.Revision != expectedRevision)
 2193                reason = $"a concurrent write advanced the ledger (revision {expectedRevision} -> {current.Revision}: a 
 2194        }
 0195        catch
 196        {
 197            // Best-effort diagnosis only — the rejection itself is what matters.
 0198        }
 199
 2200        return new InvalidOperationException(
 2201            $"Durable flow '{_flowId}' could not checkpoint because {reason}; the worker abandons this execution and the
 2202            cause);
 2203    }
 204
 205    private async Task RenewLoopAsync()
 206    {
 3207        while (!_stop.IsCancellationRequested)
 208        {
 209            try
 210            {
 3211                await Task.Delay(_options.ExecutionLeaseRenewInterval, _stop.Token).ConfigureAwait(false);
 2212                if (!await _store.TryRenewLeaseAsync(
 2213                        _flowId,
 2214                        _leaseId,
 2215                        _options.ExecutionLeaseDuration,
 2216                        _stop.Token).ConfigureAwait(false))
 217                {
 2218                    MarkLost();
 2219                    return;
 220                }
 221
 2222                _validUntilUtc = DateTime.UtcNow.Add(_options.ExecutionLeaseDuration);
 2223            }
 3224            catch (OperationCanceledException) when (_stop.IsCancellationRequested)
 225            {
 3226                return;
 227            }
 2228            catch (Exception ex)
 229            {
 2230                _logger.LogWarning(ex, "Failed to renew durable flow {FlowId} execution lease; retrying before expiry.",
 2231                if (DateTime.UtcNow >= _validUntilUtc)
 232                {
 2233                    MarkLost();
 2234                    return;
 235                }
 2236            }
 237        }
 3238    }
 239
 240    private void MarkLost()
 241    {
 242        try
 243        {
 2244            _lost.Cancel();
 2245        }
 2246        catch (ObjectDisposedException)
 247        {
 248            // Disposal won the race.
 2249        }
 2250    }
 251
 252    public async ValueTask DisposeAsync()
 253    {
 3254        if (Interlocked.Exchange(ref _disposed, 1) != 0)
 2255            return;
 256
 3257        _stop.Cancel();
 3258        await _renewal.ConfigureAwait(false);
 259
 260        try
 261        {
 3262            await _store.ReleaseLeaseAsync(_flowId, _leaseId, CancellationToken.None).ConfigureAwait(false);
 3263        }
 2264        catch (Exception ex)
 265        {
 2266            _logger.LogWarning(ex, "Failed to release durable flow {FlowId} execution lease; it will expire.", _flowId);
 2267        }
 268
 3269        _stop.Dispose();
 3270        _lost.Dispose();
 3271    }
 272}
Generated by: ReportGenerator 5.5.10.0
08/03/2026 - 08:59:44
GitHub | reportgenerator.io

Methods/Properties

.ctor(AsyncResponse.IFlowStateStore, string, string, AsyncResponse.DurableFlowOptions, Microsoft.Extensions.Logging.ILogger)
get_LostToken()
ThrowIfLost(System.Exception)
SaveAsync()
CreateSaveRejectedExceptionAsync()
RenewLoopAsync()
MarkLost()
DisposeAsync()