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

Information
Class: AsyncResponse.FlowExecutionLease
Assembly: AsyncResponse.Core
File(s): /_/src/AsyncResponse.Core/FlowStateConcurrency.cs
Line coverage
96%
Covered lines: 169
Uncovered lines: 7
Coverable lines: 176
Total lines: 628
Line coverage: 96%
Branch coverage
85%
Covered branches: 34
Total branches: 40
Branch coverage: 85%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.ctor(...)100%44100%
.cctor()100%11100%
DeadlineFrom(...)50%22100%
get_LostToken()100%11100%
get_IsLost()100%22100%
ThrowIfLost(...)100%44100%
SaveAsync()75%4496.42%
CreateSaveRejectedExceptionAsync()100%4484.61%
RenewLoopAsync()83.33%66100%
DeadlineLoopAsync()100%88100%
MarkLost()100%11100%
DisposeAsync()100%22100%
ObserveAbandonedRelease(...)25%4484%

File(s)

/_/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        EnsurePortableFlowId(flowId);
 18
 19        state.Revision = 0;
 20        return store.TryCreateAsync(flowId, state, ttl, cancellationToken);
 21    }
 22
 23    /// <summary>
 24    /// Throws <see cref="ArgumentException"/> for an id that fails <see cref="FlowIdNotPortable"/>.
 25    /// Called by every create, and by <c>IDurableFlows.StartAsync</c> BEFORE it publishes the start
 26    /// job — the publish is the start's commit point, so a job for an id no store would accept must
 27    /// never leave the process.
 28    /// </summary>
 29    internal static void EnsurePortableFlowId(string flowId)
 30    {
 31        if (FlowIdNotPortable(flowId) is { } rejection)
 32            throw new ArgumentException(rejection, nameof(flowId));
 33    }
 34
 35    /// <summary>
 36    /// Whether an existing ledger describes the same start as the requested one: same flow type,
 37    /// same input type (both ordinal), and semantically identical input JSON. The one idempotency
 38    /// test for flow ids, shared by the starter (which reports a mismatch to its caller as
 39    /// <see cref="DurableFlowIdConflictException"/>) and the executor's start target (which drops
 40    /// the job on a mismatch) so the two can never disagree about what "the same run" means.
 41    /// </summary>
 42    internal static bool IsSameStart(FlowState existing, string? flowTypeName, string? inputTypeName, string? inputJson)
 43        => string.Equals(existing.FlowTypeName, flowTypeName, StringComparison.Ordinal)
 44            && string.Equals(existing.InputTypeName, inputTypeName, StringComparison.Ordinal)
 45            && FlowStateJson.JsonEquivalent(existing.InputJson, inputJson ?? string.Empty);
 46
 47    /// <summary>
 48    /// Enforces the portable flow-id contract on every final id at creation — the single door all
 49    /// creates walk through. Three independent limits, because the stores disagree about what an
 50    /// id may be, and an id that works on one store and fails on another is not portable:
 51    /// <list type="bullet">
 52    /// <item>length in UTF-16 code units, for the 400-unit <c>flow_id</c> columns (SQL Server,
 53    /// MySQL, Oracle, EF Core);</item>
 54    /// <item>length in UTF-8 <em>bytes</em>, for Cosmos DB, whose 1023-byte id limit a 400-unit id
 55    /// exceeds once the characters are non-ASCII (up to three bytes per unit, four for a
 56    /// surrogate pair);</item>
 57    /// <item>the characters themselves — Cosmos rejects <c>/</c>, <c>\</c>, <c>?</c> and <c>#</c>
 58    /// in an id, and control characters break every store's diagnostics;</item>
 59    /// <item>no surrounding spaces — SQL Server pads the shorter operand of an equality
 60    /// comparison (binary collations included) and MySQL's <c>utf8mb4_bin</c> is PAD SPACE, so
 61    /// <c>flow</c> and <c>flow&#160;</c> are ONE key to those databases while the engine treats
 62    /// them as two runs.</item>
 63    /// </list>
 64    /// Case is deliberately NOT folded here: ids are compared ordinally throughout, and the
 65    /// relational stores pin a binary collation on the column so the database agrees.
 66    /// Returns the rejection message, or <c>null</c> when the id is portable.
 67    /// </summary>
 68    internal static string? FlowIdNotPortable(string flowId)
 69    {
 70        // Length-guarded before the [0]/[^1] probe below: this is the single door every create
 71        // walks through and its contract is to RETURN a rejection, so an empty id must not throw
 72        // IndexOutOfRangeException out of the very method whose job is to explain bad ids.
 73        if (flowId.Length == 0)
 74            return "Flow id is empty. A run needs an id to be addressable by its wake-ups, child flows, and recovery cal
 75
 76        if (flowId.Length > DurableFlowOptions.MaxFlowIdLength)
 77        {
 78            return $"Flow id '{Excerpt(flowId)}' is {flowId.Length} UTF-16 code units; the portable maximum is " +
 79                $"{DurableFlowOptions.MaxFlowIdLength} ({nameof(DurableFlowOptions)}.{nameof(DurableFlowOptions.MaxFlowI
 80                "column length in the SQL Server, MySQL, Oracle, and EF Core stores). " + BudgetGuidance;
 81        }
 82
 83        // Checked BEFORE the byte count, which would otherwise be measured against the U+FFFD an
 84        // encoder substitutes rather than against the id the caller passed.
 85        if (PortableText.IndexOfIllFormedUtf16(flowId) is var illFormed and >= 0)
 86            return PortableText.IllFormedUtf16Rejection("Flow id", Excerpt(flowId), flowId[illFormed], illFormed);
 87
 88        var utf8Bytes = System.Text.Encoding.UTF8.GetByteCount(flowId);
 89        if (utf8Bytes > DurableFlowOptions.MaxFlowIdBytes)
 90        {
 91            return $"Flow id '{Excerpt(flowId)}' is {utf8Bytes} UTF-8 bytes; the portable maximum is " +
 92                $"{DurableFlowOptions.MaxFlowIdBytes} ({nameof(DurableFlowOptions)}.{nameof(DurableFlowOptions.MaxFlowId
 93                "id limit). A non-ASCII character costs up to three bytes (four for a surrogate pair), so a count of cha
 94                "does not bound the byte length. " + BudgetGuidance;
 95        }
 96
 97        if (flowId[0] == ' ' || flowId[^1] == ' ')
 98        {
 99            return $"Flow id '{Excerpt(flowId)}' begins or ends with a space. SQL Server pads the shorter operand of an 
 100                "comparison — binary collations included — and MySQL's utf8mb4_bin is PAD SPACE, so an id with trailing 
 101                "the SAME key as one without to those stores, while the engine compares them ordinally and treats them a
 102                "different flows. Trim the id.";
 103        }
 104
 105        foreach (var character in flowId)
 106        {
 107            if (character is '/' or '\\' or '?' or '#' || char.IsControl(character))
 108            {
 109                return $"Flow id '{Excerpt(flowId)}' contains the character '{(char.IsControl(character) ? $"\\u{(int)ch
 110                    "which is not portable: Cosmos DB rejects '/', '\\', '?' and '#' in an id, and control characters co
 111                    "diagnostics. Use a separator the stores agree on, such as ':' or '-'.";
 112            }
 113        }
 114
 115        return null;
 116    }
 117
 118    private const string BudgetGuidance =
 119        "Budget root ids for growth: child flows append \":{stepName}\" to the parent id, and scheduled flows wrap the s
 120        "name as \"sched:{name}:{timestamp}\".";
 121
 122    private static string Excerpt(string flowId) => PortableText.Excerpt(flowId);
 123
 124    public static async Task<FlowExecutionLease?> TryAcquireExecutionLeaseAsync(
 125        IFlowStateStore store,
 126        string flowId,
 127        DurableFlowOptions options,
 128        ILogger logger,
 129        TimeProvider? timeProvider = null,
 130        CancellationToken cancellationToken = default,
 131        string? jobTag = null)
 132    {
 133        ValidateOptions(options);
 134
 135        var clock = timeProvider ?? TimeProvider.System;
 136
 137        // The job driving this execution is recorded IN the lease id, so it lands atomically with
 138        // the acquire on every store and comes back through ObserveLeaseAsync: a later delivery of
 139        // that same job can then tell it is contending with its own first delivery (see
 140        // FlowLeaseContention). Without a job identity this is the plain 32-character id.
 141        var leaseId = FlowLeaseContention.NewLeaseId(jobTag);
 142
 143        // Stamp the deadline BEFORE the call, not after it returns. The store starts the lease when
 144        // it executes the command; every millisecond after that — network latency, a delayed
 145        // continuation, a GC pause between the response arriving and this line running — is lease
 146        // time already spent. Anchoring afterwards handed that whole interval back to the client as
 147        // if it were still owned, so a worker could believe it held a 60s lease 20s past the point
 148        // another replica was free to take it. Anchoring first is conservative in the safe
 149        // direction: the client's deadline can only be EARLIER than the server's.
 150        var deadline = FlowExecutionLease.DeadlineFrom(clock, options.ExecutionLeaseDuration);
 151
 152        if (!await store.TryAcquireLeaseAsync(
 153                flowId,
 154                leaseId,
 155                options.ExecutionLeaseDuration,
 156                cancellationToken).ConfigureAwait(false))
 157            return null;
 158
 159        // The constructor is throw-free after the option bounds above: it only assigns fields,
 160        // records the pre-call deadline, and starts the renewal loop (whose first Task.Delay faults
 161        // the loop task, never the constructor). Were that ever to change, lease expiry is the
 162        // backstop for the persisted row.
 163        return new FlowExecutionLease(store, flowId, leaseId, options, logger, clock, deadline);
 164    }
 165
 166    public static async Task<bool> MutateAsync(
 167        IFlowStateStore store,
 168        string flowId,
 169        TimeSpan ttl,
 170        TimeProvider? timeProvider,
 171        Func<FlowState, bool> mutate,
 172        CancellationToken cancellationToken = default)
 173    {
 174        for (var attempt = 0; attempt < MaxUpdateAttempts; attempt++)
 175        {
 176            var state = await store.LoadAsync(flowId, cancellationToken).ConfigureAwait(false);
 177            if (state is null)
 178                return false;
 179
 180            if (!mutate(state))
 181                return true;
 182
 183            var expectedRevision = state.Revision;
 184            state.Revision = checked(expectedRevision + 1);
 185            var nowUtc = (timeProvider ?? TimeProvider.System).GetUtcNow().UtcDateTime;
 186            state.UpdatedAtUtc = nowUtc;
 187            if (await store.TryUpdateAsync(
 188                    flowId,
 189                    state,
 190                    expectedRevision,
 191                    // A lease-bypassing write (recovery, failure signal, operator) never shrinks a
 192                    // live run's ledger under a park it knows nothing about.
 193                    FlowStateRetention.EffectiveTtl(state, ttl, nowUtc),
 194                    leaseId: null,
 195                    cancellationToken).ConfigureAwait(false))
 196                return true;
 197        }
 198
 199        throw new InvalidOperationException(
 200            $"Durable flow '{flowId}' changed repeatedly while applying a recovery update; retry the operation.");
 201    }
 202
 203    internal static void ValidateOptions(DurableFlowOptions options)
 204    {
 205        // Upper bounds close the "passes validation, throws mid-operation" gap — but only on the
 206        // knobs that actually reach the failing sink. StateExpiry and ExecutionLeaseDuration become
 207        // "now + value" stamps (store TTLs and lease deadlines) and never arm a timer themselves,
 208        // so they get the persistence bound; DefaultStepTimeout and ExecutionLeaseRenewInterval arm
 209        // BCL timers, so they get the timer ceiling; ProgressPersistenceInterval is only ever
 210        // compared against elapsed time (DurableFlowContext.ReportProgressAsync), so any
 211        // non-negative value is representable — a 60-day lease or progress throttle is a valid
 212        // configuration and must not fail startup.
 213        AsyncResponseChannelOptions.EnsurePersistedTtl(options.StateExpiry, nameof(DurableFlowOptions), nameof(options.S
 214        if (options.DefaultStepTimeout is { } defaultStepTimeout)
 215            AsyncResponseChannelOptions.EnsureTimerBacked(defaultStepTimeout, nameof(DurableFlowOptions), nameof(options
 216        AsyncResponseChannelOptions.EnsurePersistedTtl(options.ExecutionLeaseDuration, nameof(DurableFlowOptions), nameo
 217        AsyncResponseChannelOptions.EnsureTimerBacked(options.ExecutionLeaseRenewInterval, nameof(DurableFlowOptions), n
 218        if (options.LedgerSizeWarningBytes is { } ledgerWarning && ledgerWarning <= 0)
 219        {
 220            throw new InvalidOperationException(
 221                $"{nameof(DurableFlowOptions)}.{nameof(options.LedgerSizeWarningBytes)} must be positive, or null to dis
 222        }
 223        if (options.MaxRetainedSteps is <= 0)
 224            throw new InvalidOperationException($"{nameof(DurableFlowOptions)}.{nameof(options.MaxRetainedSteps)} must b
 225        if (options.ExecutionLeaseRenewInterval >= options.ExecutionLeaseDuration)
 226        {
 227            throw new InvalidOperationException(
 228                $"{nameof(DurableFlowOptions)}.{nameof(options.ExecutionLeaseRenewInterval)} must be shorter than " +
 229                $"{nameof(DurableFlowOptions.ExecutionLeaseDuration)}.");
 230        }
 231        // Compared against elapsed time only (the contention poll arms pollDelay-sized timers), so
 232        // any positive value is representable; zero or less would cap the store-driven wait below
 233        // "no wait at all", which is a misconfiguration rather than a way to disable the feature.
 234        if (options.MaxLeaseContentionWait <= TimeSpan.Zero)
 235            throw new InvalidOperationException($"{nameof(DurableFlowOptions)}.{nameof(options.MaxLeaseContentionWait)} 
 236        if (options.ProgressPersistenceInterval < TimeSpan.Zero)
 237            throw new InvalidOperationException($"{nameof(DurableFlowOptions)}.{nameof(options.ProgressPersistenceInterv
 238        // Timer remainders at or under the threshold arm an in-process Task.Delay, so the knob is
 239        // timer-backed; zero legitimately means "always suspend".
 240        AsyncResponseChannelOptions.EnsureTimerBackedAllowZero(options.TimerInProcessThreshold, nameof(DurableFlowOption
 241    }
 242}
 243
 244/// <summary>One distributed durable-flow execution lease.</summary>
 245internal sealed class FlowExecutionLease : IAsyncDisposable
 246{
 247    private readonly IFlowStateStore _store;
 248    private readonly string _flowId;
 249    private readonly string _leaseId;
 250    private readonly DurableFlowOptions _options;
 251    private readonly ILogger _logger;
 252    private readonly TimeProvider _timeProvider;
 2116253    private readonly CancellationTokenSource _stop = new();
 2116254    private readonly CancellationTokenSource _lost = new();
 255    private readonly Task _renewal;
 256    private readonly Task _deadline;
 257    // DateTime ticks so the renewal loop's writes and the execution path's reads tear-free on
 258    // 32-bit runtimes and order via Volatile.
 259    private long _validUntilUtcTicks;
 260    private int _disposed;
 261
 262    /// <summary>
 263    /// Longest single wait the deadline watcher arms. ExecutionLeaseDuration is validated as a
 264    /// PERSISTENCE bound, not a timer bound (see <see cref="FlowStateConcurrency.ValidateOptions"/>) —
 265    /// a 60-day lease is a legal configuration — so the watcher sleeps in chunks and re-reads the
 266    /// deadline rather than handing an out-of-range delay to a BCL timer.
 267    /// </summary>
 13268    private static readonly TimeSpan MaxDeadlineChunk = TimeSpan.FromDays(1);
 269
 270    /// <summary>
 271    /// Budget for joining the renewal and deadline loops on disposal. The renewal loop can be
 272    /// stuck inside a store call that ignores its cancellation token (a wedged connection, a
 273    /// database that accepts the request and never answers — the exact case the deadline watcher
 274    /// exists for); an unbounded join there wedged the whole worker job. Past the budget the
 275    /// loops are abandoned: the deadline watcher has already marked the lease lost and the
 276    /// server-side lease expires on its own.
 277    /// </summary>
 13278    private static readonly TimeSpan DisposeJoinLimit = TimeSpan.FromSeconds(30);
 279
 280    /// <summary>
 281    /// Budget for the final lease release on disposal. The release is one conditional write, so
 282    /// ten seconds is generous; past it the call is abandoned (cancelled, its outcome observed)
 283    /// and the server-side lease expires on its own — the same recovery the abandoned renewal
 284    /// loops rely on. Separate from <see cref="DisposeJoinLimit"/> because the two hang for
 285    /// different reasons: the loops are joined first and are usually idle, while the release is
 286    /// a fresh store call that a wedged connection can hold indefinitely even after a clean join.
 287    /// </summary>
 13288    private static readonly TimeSpan ReleaseLimit = TimeSpan.FromSeconds(10);
 289
 290    /// <param name="store">The flow state store the lease was acquired through.</param>
 291    /// <param name="flowId">The flow the lease protects.</param>
 292    /// <param name="leaseId">The identity of this lease within the flow's row.</param>
 293    /// <param name="options">Validated durable-flow options.</param>
 294    /// <param name="logger">Sink for renewal and deadline watcher events.</param>
 295    /// <param name="timeProvider">Clock used for deadline computation; <see cref="TimeProvider.System"/> when omitted.<
 296    /// <param name="acquiredDeadlineUtcTicks">
 297    /// The conservative deadline for the lease this instance was handed, captured BEFORE the
 298    /// acquire call went out. Omitted only by callers that construct a lease without an acquire
 299    /// round trip (tests), where "now + duration" is exact.
 300    /// </param>
 2116301    public FlowExecutionLease(
 2116302        IFlowStateStore store,
 2116303        string flowId,
 2116304        string leaseId,
 2116305        DurableFlowOptions options,
 2116306        ILogger logger,
 2116307        TimeProvider? timeProvider = null,
 2116308        long? acquiredDeadlineUtcTicks = null)
 309    {
 2116310        _store = store;
 2116311        _flowId = flowId;
 2116312        _leaseId = leaseId;
 2116313        _options = options;
 2116314        _logger = logger;
 2116315        _timeProvider = timeProvider ?? TimeProvider.System;
 2116316        Volatile.Write(
 2116317            ref _validUntilUtcTicks,
 2116318            acquiredDeadlineUtcTicks ?? DeadlineFrom(_timeProvider, options.ExecutionLeaseDuration));
 2116319        _renewal = RenewLoopAsync();
 2116320        _deadline = DeadlineLoopAsync();
 2116321    }
 322
 323    /// <summary>
 324    /// "Now + duration" in UTC ticks, saturating instead of overflowing: <c>ExecutionLeaseDuration</c>
 325    /// is bounded as a persistence TTL, not a timer, so a 60-day lease near <see cref="DateTime.MaxValue"/>
 326    /// is a legal configuration that must not throw here.
 327    /// </summary>
 328    internal static long DeadlineFrom(TimeProvider timeProvider, TimeSpan duration)
 329    {
 8067330        var now = timeProvider.GetUtcNow().UtcDateTime;
 8067331        return duration > DateTime.MaxValue - now ? DateTime.MaxValue.Ticks : now.Add(duration).Ticks;
 332    }
 333
 1685334    public CancellationToken LostToken => _lost.Token;
 335
 336    /// <summary>
 337    /// Whether this lease can still fence a write. Callers holding a claimed, unrecorded result
 338    /// use it to choose the lease-less persistence path BEFORE surfacing the takeover signal.
 339    /// </summary>
 14340    public bool IsLost => _lost.IsCancellationRequested
 14341        || _timeProvider.GetUtcNow().UtcDateTime.Ticks >= Volatile.Read(ref _validUntilUtcTicks);
 342
 343    /// <summary>
 344    /// Throws when the lease is lost. <paramref name="cause"/> (e.g. the exception that made the
 345    /// caller check) is attached as the inner exception so the real failure is not discarded.
 346    /// <para>
 347    /// A passed deadline counts as lost even before any renewal fails: the renewal loop only
 348    /// observes loss on a store round-trip, so a stop-the-world pause (GC, VM freeze, debugger)
 349    /// longer than the lease lets another worker take over while this side has seen nothing —
 350    /// its next step body would then run concurrently with the new holder's. Checkpoints are
 351    /// lease-fenced; side effects are fenced only by this guard, so it is conservative near the
 352    /// boundary by design: retrying from the checkpoint is always safe, a concurrent step is not.
 353    /// </para>
 354    /// </summary>
 355    public void ThrowIfLost(Exception? cause = null)
 356    {
 23366357        if (!_lost.IsCancellationRequested
 23366358            && _timeProvider.GetUtcNow().UtcDateTime.Ticks < Volatile.Read(ref _validUntilUtcTicks))
 23348359            return;
 360
 18361        MarkLost();
 18362        throw new InvalidOperationException($"Durable flow '{_flowId}' lost its execution lease; the worker will retry f
 363    }
 364
 365    public async Task SaveAsync(FlowState state, TimeSpan ttl, CancellationToken cancellationToken = default, Exception?
 366    {
 12354367        ThrowIfLost(cause);
 12350368        var expectedRevision = state.Revision;
 12350369        state.Revision = checked(expectedRevision + 1);
 12350370        var nowUtc = _timeProvider.GetUtcNow().UtcDateTime;
 12350371        state.UpdatedAtUtc = nowUtc;
 372
 373        try
 374        {
 12350375            if (await _store.TryUpdateAsync(
 12350376                    _flowId,
 12350377                    state,
 12350378                    expectedRevision,
 12350379                    // Every checkpoint carries the ledger's retention floor forward (see
 12350380                    // FlowStateRetention): the executor's per-attempt save and an ancestor's
 12350381                    // re-park stamp the plain StateExpiry, and used to shrink a ledger a
 12350382                    // descendant had extended for a wait still in progress.
 12350383                    FlowStateRetention.EffectiveTtl(state, ttl, nowUtc),
 12350384                    _leaseId,
 12350385                    cancellationToken).ConfigureAwait(false))
 12318386                return;
 28387        }
 4388        catch
 389        {
 4390            state.Revision = expectedRevision;
 4391            MarkLost();
 392
 393            // The store exception propagates; keep the failure this save was recording from
 394            // vanishing with it.
 4395            if (cause is not null)
 0396                _logger.LogWarning(cause, "Durable flow '{FlowId}' failed to checkpoint; the failure it was recording is
 4397            throw;
 398        }
 399
 28400        state.Revision = expectedRevision;
 28401        MarkLost();
 28402        throw await CreateSaveRejectedExceptionAsync(expectedRevision, cause, cancellationToken).ConfigureAwait(false);
 12318403    }
 404
 405    /// <summary>
 406    /// Builds the exception for a rejected checkpoint write. The store's compare-and-swap only
 407    /// returns <c>false</c>, so the reason is diagnosed with a best-effort re-read: a revision
 408    /// conflict — a concurrent lease-bypassing writer such as <c>RecoverAsync</c>, <c>FailAsync</c>,
 409    /// or an operator parking the run — is reported as such instead of as a lost lease, which sent
 410    /// operators hunting phantom lease problems. Behavior is unchanged either way: the lease is
 411    /// abandoned (<see cref="MarkLost"/> already ran) and the delivery retries from the last
 412    /// checkpoint; <paramref name="cause"/> rides along as the inner exception so the failure that
 413    /// triggered the save is not discarded.
 414    /// </summary>
 415    private async Task<InvalidOperationException> CreateSaveRejectedExceptionAsync(
 416        long expectedRevision,
 417        Exception? cause,
 418        CancellationToken cancellationToken)
 419    {
 28420        var reason = "its execution lease was no longer held (expired or taken over)";
 421        try
 422        {
 28423            var current = await _store.LoadAsync(_flowId, cancellationToken).ConfigureAwait(false);
 28424            if (current is null)
 6425                reason = "its ledger entry is gone (expired or deleted)";
 22426            else if (current.Revision != expectedRevision)
 10427                reason = $"a concurrent write advanced the ledger (revision {expectedRevision} -> {current.Revision}: a 
 28428        }
 0429        catch
 430        {
 431            // Best-effort diagnosis only — the rejection itself is what matters.
 0432        }
 433
 28434        return new InvalidOperationException(
 28435            $"Durable flow '{_flowId}' could not checkpoint because {reason}; the worker abandons this execution and the
 28436            cause);
 28437    }
 438
 439    private async Task RenewLoopAsync()
 440    {
 3851441        while (!_stop.IsCancellationRequested)
 442        {
 443            try
 444            {
 3851445                await Task.Delay(_options.ExecutionLeaseRenewInterval, _timeProvider, _stop.Token).ConfigureAwait(false)
 446
 447                // Same anchoring rule as acquisition: the renewed lease starts when the store runs
 448                // the command, so the deadline is measured from before the call, not from whenever
 449                // the answer gets back here. Published only on success, so a failed renewal never
 450                // extends anything.
 1746451                var renewedDeadline = DeadlineFrom(_timeProvider, _options.ExecutionLeaseDuration);
 452
 1746453                if (!await _store.TryRenewLeaseAsync(
 1746454                        _flowId,
 1746455                        _leaseId,
 1746456                        _options.ExecutionLeaseDuration,
 1746457                        _stop.Token).ConfigureAwait(false))
 458                {
 6459                    MarkLost();
 6460                    return;
 461                }
 462
 1728463                Volatile.Write(ref _validUntilUtcTicks, renewedDeadline);
 1728464            }
 2083465            catch (OperationCanceledException) when (_stop.IsCancellationRequested)
 466            {
 2083467                return;
 468            }
 8469            catch (Exception ex)
 470            {
 8471                _logger.LogWarning(ex, "Failed to renew durable flow {FlowId} execution lease; retrying before expiry.",
 8472                if (_timeProvider.GetUtcNow().UtcDateTime.Ticks >= Volatile.Read(ref _validUntilUtcTicks))
 473                {
 1474                    MarkLost();
 1475                    return;
 476                }
 7477            }
 478        }
 2090479    }
 480
 481    /// <summary>
 482    /// Cancels <see cref="LostToken"/> when the lease deadline passes, on a clock of its own.
 483    /// <para>
 484    /// <see cref="RenewLoopAsync"/> cannot be trusted to do this: it only learns the lease is gone
 485    /// by completing a store round-trip, so a renewal call that hangs — a wedged connection, a
 486    /// database that accepts the request and never answers — leaves the token live indefinitely
 487    /// while the server-side lease expires and another replica takes the flow over. Checkpoints
 488    /// stay fenced regardless (<see cref="ThrowIfLost"/> and the lease-fenced CAS both check the
 489    /// clock), but anything watching the TOKEN — a step body, a linked operation — saw nothing.
 490    /// This loop closes that gap: it re-reads the deadline each pass, so a successful renewal
 491    /// simply pushes it out, and it fires whether or not the renewal path is responsive.
 492    /// </para>
 493    /// </summary>
 494    private async Task DeadlineLoopAsync()
 495    {
 496        try
 497        {
 27660498            while (!_stop.IsCancellationRequested && !_lost.IsCancellationRequested)
 499            {
 27656500                var remaining = new DateTime(Volatile.Read(ref _validUntilUtcTicks), DateTimeKind.Utc)
 27656501                    - _timeProvider.GetUtcNow().UtcDateTime;
 502
 27656503                if (remaining <= TimeSpan.Zero)
 504                {
 6505                    _logger.LogWarning(
 6506                        "Durable flow {FlowId} execution lease reached its deadline without a successful renewal; abando
 6507                        _flowId);
 6508                    MarkLost();
 6509                    return;
 510                }
 511
 27650512                await Task.Delay(
 27650513                    remaining < MaxDeadlineChunk ? remaining : MaxDeadlineChunk,
 27650514                    _timeProvider,
 27650515                    _stop.Token).ConfigureAwait(false);
 516            }
 4517        }
 2082518        catch (OperationCanceledException) when (_stop.IsCancellationRequested)
 519        {
 520            // Normal completion: the execution finished and disposal stopped the watcher.
 2082521        }
 2092522    }
 523
 524    private void MarkLost()
 525    {
 526        try
 527        {
 65528            _lost.Cancel();
 63529        }
 2530        catch (ObjectDisposedException)
 531        {
 532            // Disposal won the race.
 2533        }
 65534    }
 535
 536    public async ValueTask DisposeAsync()
 537    {
 2092538        if (Interlocked.Exchange(ref _disposed, 1) != 0)
 2539            return;
 540
 2090541        _stop.Cancel();
 542        try
 543        {
 544            // Bounded join (see DisposeJoinLimit): both loops swallow their own exceptions, so an
 545            // abandoned task cannot fault unobserved.
 2090546            await Task.WhenAll(_renewal, _deadline).WaitAsync(DisposeJoinLimit, _timeProvider).ConfigureAwait(false);
 2087547        }
 3548        catch (TimeoutException)
 549        {
 550            // The store call the renewal loop is stuck in ignores cancellation, so releasing the
 551            // lease through the same store would hang this disposal all over again. Skip the
 552            // release (the server-side lease expires) and leave the cancellation sources
 553            // undisposed for the abandoned loops.
 3554            _logger.LogWarning(
 3555                "Durable flow {FlowId} execution lease loops did not stop within {DisposeJoinLimit}; abandoning them (th
 3556                _flowId,
 3557                DisposeJoinLimit);
 3558            return;
 559        }
 560
 561        // Bounded release (see ReleaseLimit), with a token the store can honor. An unbounded,
 562        // uncancelable release kept a FINISHED execution's disposal — and with it the executor's
 563        // `await using`, the job's DI scope, the worker slot, and the transport acknowledgement —
 564        // pending for as long as a wedged store took to answer, which can be forever.
 2087565        var releaseCancellation = new CancellationTokenSource();
 2087566        Task? release = null;
 567        try
 568        {
 2087569            release = _store.ReleaseLeaseAsync(_flowId, _leaseId, releaseCancellation.Token);
 2087570            await release.WaitAsync(ReleaseLimit, _timeProvider).ConfigureAwait(false);
 2082571            releaseCancellation.Dispose();
 2082572        }
 2573        catch (TimeoutException) when (release is { IsCompleted: false })
 574        {
 575            // The budget lapsed with the store still silent (a TimeoutException thrown BY the
 576            // store completes the task first and takes the branch below). Cancel what can be
 577            // cancelled, observe whatever the abandoned call eventually does, and move on: the
 578            // server-side lease expires on its own, exactly as when the renewal loops are abandoned.
 2579            releaseCancellation.Cancel();
 2580            _logger.LogWarning(
 2581                "Durable flow {FlowId} execution lease release did not complete within {ReleaseLimit}; abandoning it (th
 2582                _flowId,
 2583                ReleaseLimit);
 2584            ObserveAbandonedRelease(release, releaseCancellation);
 2585        }
 3586        catch (Exception ex)
 587        {
 3588            releaseCancellation.Dispose();
 3589            _logger.LogWarning(ex, "Failed to release durable flow {FlowId} execution lease; it will expire.", _flowId);
 3590        }
 591
 2087592        _stop.Dispose();
 2087593        _lost.Dispose();
 2092594    }
 595
 596    /// <summary>
 597    /// Attaches the one continuation an abandoned release needs: its eventual fault is observed
 598    /// (and logged, so a store that finally answers with an error is not an unobserved-task
 599    /// event) and the cancellation source it still holds is disposed only once it can no longer
 600    /// be touched.
 601    /// </summary>
 602    private void ObserveAbandonedRelease(Task release, CancellationTokenSource releaseCancellation)
 2603        => _ = release.ContinueWith(
 2604            (task, state) =>
 2605            {
 2606                var (lease, cancellation) = ((FlowExecutionLease, CancellationTokenSource))state!;
 2607                if (task.IsFaulted)
 2608                {
 0609                    lease._logger.LogWarning(
 0610                        task.Exception?.GetBaseException(),
 0611                        "The abandoned release of durable flow {FlowId}'s execution lease eventually failed; the lease e
 0612                        lease._flowId);
 2613                }
 2614                else
 2615                {
 2616                    lease._logger.LogDebug(
 2617                        "The abandoned release of durable flow {FlowId}'s execution lease eventually completed ({Status}
 2618                        lease._flowId,
 2619                        task.Status);
 2620                }
 2621
 2622                cancellation.Dispose();
 2623            },
 2624            (this, releaseCancellation),
 2625            CancellationToken.None,
 2626            TaskContinuationOptions.ExecuteSynchronously,
 2627            TaskScheduler.Default);
 628}
Generated by: ReportGenerator 5.5.11.0
09/22/2026 - 09:41:04
GitHub | reportgenerator.io

Methods/Properties

.ctor(AsyncResponse.IFlowStateStore,System.String,System.String,AsyncResponse.DurableFlowOptions,Microsoft.Extensions.Logging.ILogger,System.TimeProvider,System.Nullable`1<System.Int64>)
.cctor()
DeadlineFrom(System.TimeProvider,System.TimeSpan)
get_LostToken()
get_IsLost()
ThrowIfLost(System.Exception)
SaveAsync()
CreateSaveRejectedExceptionAsync()
RenewLoopAsync()
DeadlineLoopAsync()
MarkLost()
DisposeAsync()
ObserveAbandonedRelease(System.Threading.Tasks.Task,System.Threading.CancellationTokenSource)