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

Information
Class: AsyncResponse.InMemoryFlowStateStore
Assembly: AsyncResponse.Core
File(s): /_/src/AsyncResponse.Core/InMemoryFlowStateStore.cs
Line coverage
99%
Covered lines: 124
Uncovered lines: 1
Coverable lines: 125
Total lines: 297
Line coverage: 99.2%
Branch coverage
90%
Covered branches: 87
Total branches: 96
Branch coverage: 90.6%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
Expiry(...)100%22100%
.cctor()100%11100%
.ctor(...)100%22100%
TryCreateAsync(...)80%1010100%
LoadAsync(...)100%88100%
TryUpdateAsync(...)90%2020100%
TryAcquireLeaseAsync(...)100%11100%
TryRenewLeaseAsync(...)100%11100%
ReleaseLeaseAsync(...)83.33%66100%
ObserveLeaseAsync(...)100%44100%
ExpireAllLeases()100%88100%
TryDeleteAsync(...)100%11100%
SweepExpired(...)87.5%8890%
TryChangeLeaseAsync(...)86.36%2222100%
CreateEntry(...)100%11100%
ValidateWrite(...)100%66100%
.ctor(...)100%11100%
get_StateJson()100%11100%
get_Revision()100%11100%
get_ExpiresAtUtc()100%11100%
get_LeaseId()100%11100%
get_LeaseExpiresAtUtc()100%11100%

File(s)

/_/src/AsyncResponse.Core/InMemoryFlowStateStore.cs

#LineLine coverage
 1using System.Collections.Concurrent;
 2
 3namespace AsyncResponse;
 4
 5/// <summary>Atomic process-local flow-state store for development, tests, and single-process apps.</summary>
 6internal sealed class InMemoryFlowStateStore : IFlowStateStore
 7{
 8    // Saturating expiry stamp: the ttl parameter arrives from callers as well as options, and the
 9    // external stores deliberately saturate the same arithmetic — a raw Add threw
 10    // ArgumentOutOfRangeException on large ttls where every other store clamped.
 11    private static DateTime Expiry(DateTime now, TimeSpan ttl)
 944212        => ttl >= DateTime.MaxValue - now ? DateTime.MaxValue : now.Add(ttl);
 13
 14    /// <summary>
 15    /// How often <see cref="TryCreateAsync"/> sweeps expired entries whose ids are never touched
 16    /// again. Expiry used to be enforced only on access to the SAME id (a load or a replacement),
 17    /// which a completed run normally never gets — so a long-lived process minting unique flow
 18    /// ids retained every expired ledger (inputs, memoized results, value bags) for its lifetime;
 19    /// <c>StateExpiry</c> hid them from reads without ever bounding memory. On the engine's clock,
 20    /// like every other stamp here: a virtual clock that never advances never sweeps, and
 21    /// nothing has expired under it either.
 22    /// </summary>
 723    internal static readonly TimeSpan SweepInterval = TimeSpan.FromMinutes(1);
 24
 91425    private readonly ConcurrentDictionary<string, Entry> _entries = new(StringComparer.Ordinal);
 26    private readonly TimeProvider _timeProvider;
 27    private long _nextSweepTicks;
 28
 29    /// <summary>Creates the store; expiry and lease stamps come from the engine's clock.</summary>
 91430    public InMemoryFlowStateStore(TimeProvider? timeProvider = null)
 91431        => _timeProvider = timeProvider ?? TimeProvider.System;
 32
 33    public Task<bool> TryCreateAsync(
 34        string flowId,
 35        FlowState state,
 36        TimeSpan ttl,
 37        CancellationToken cancellationToken = default)
 38    {
 370639        ValidateWrite(flowId, state, ttl);
 369640        cancellationToken.ThrowIfCancellationRequested();
 369441        if (state.Revision != 0)
 242            throw new ArgumentException("A new flow ledger must start at revision zero.", nameof(state));
 43
 369244        var now = _timeProvider.GetUtcNow().UtcDateTime;
 369245        SweepExpired(now);
 369246        var created = CreateEntry(state, Expiry(now, ttl));
 47        while (true)
 48        {
 369249            if (_entries.TryAdd(flowId, created))
 330850                return Task.FromResult(true);
 51
 38452            if (!_entries.TryGetValue(flowId, out var existing))
 53                continue;
 54
 38455            if (existing.ExpiresAtUtc > now)
 38256                return Task.FromResult(false);
 57
 258            if (_entries.TryUpdate(flowId, created, existing))
 259                return Task.FromResult(true);
 60        }
 61    }
 62
 63    public Task<FlowState?> LoadAsync(string flowId, CancellationToken cancellationToken = default)
 64    {
 808765        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 808566        cancellationToken.ThrowIfCancellationRequested();
 67
 809068        while (_entries.TryGetValue(flowId, out var entry))
 69        {
 791870            if (entry.ExpiresAtUtc <= _timeProvider.GetUtcNow().UtcDateTime)
 71            {
 672                _entries.TryRemove(KeyValuePair.Create(flowId, entry));
 673                continue;
 74            }
 75
 76            // Unreadable JSON, an unknown schema version, and an entry whose JSON disagrees with
 77            // its own revision or key all throw out of here rather than masquerading as a deleted
 78            // flow: the entry is present, so acknowledging its wake-up as "gone" would strand the
 79            // run. Same contract as the durable stores — see DurableFlowStoreShared.ReadState.
 791280            var state = FlowStateJson.Deserialize(entry.StateJson, flowId);
 790881            if (state.Revision != entry.Revision)
 82            {
 683                throw new FlowStateUnreadableException(
 684                    flowId,
 685                    $"its stored revision is {entry.Revision} but the revision inside its JSON is {state.Revision}");
 86            }
 87
 790288            if (!string.Equals(state.FlowId, flowId, StringComparison.Ordinal))
 289                throw new FlowStateUnreadableException(flowId, "the flow id inside its JSON is not the id it is stored u
 90
 790091            return Task.FromResult<FlowState?>(state);
 92        }
 93
 17294        return Task.FromResult<FlowState?>(null);
 95    }
 96
 97    public Task<bool> TryUpdateAsync(
 98        string flowId,
 99        FlowState state,
 100        long expectedRevision,
 101        TimeSpan ttl,
 102        string? leaseId = null,
 103        CancellationToken cancellationToken = default)
 104    {
 5212105        ValidateWrite(flowId, state, ttl);
 5212106        cancellationToken.ThrowIfCancellationRequested();
 5212107        if (expectedRevision < 0)
 4108            throw new ArgumentOutOfRangeException(nameof(expectedRevision), "Expected revision cannot be negative.");
 5208109        if (state.Revision != checked(expectedRevision + 1))
 2110            throw new ArgumentException("The new flow-state revision must increment the expected revision by one.", name
 111
 5206112        while (_entries.TryGetValue(flowId, out var current))
 113        {
 5198114            var now = _timeProvider.GetUtcNow().UtcDateTime;
 5198115            if (current.ExpiresAtUtc <= now || current.Revision != expectedRevision)
 22116                return Task.FromResult(false);
 5176117            if (leaseId is not null
 5176118                && (!string.Equals(current.LeaseId, leaseId, StringComparison.Ordinal)
 5176119                    || current.LeaseExpiresAtUtc <= now))
 16120                return Task.FromResult(false);
 121
 5160122            var updated = CreateEntry(
 5160123                state,
 5160124                Expiry(now, ttl),
 5160125                current.LeaseId,
 5160126                current.LeaseExpiresAtUtc);
 5160127            if (_entries.TryUpdate(flowId, updated, current))
 5160128                return Task.FromResult(true);
 129        }
 130
 8131        return Task.FromResult(false);
 132    }
 133
 134    public Task<bool> TryAcquireLeaseAsync(
 135        string flowId,
 136        string leaseId,
 137        TimeSpan leaseDuration,
 138        CancellationToken cancellationToken = default)
 1278139        => TryChangeLeaseAsync(flowId, leaseId, leaseDuration, acquire: true, cancellationToken);
 140
 141    public Task<bool> TryRenewLeaseAsync(
 142        string flowId,
 143        string leaseId,
 144        TimeSpan leaseDuration,
 145        CancellationToken cancellationToken = default)
 1739146        => TryChangeLeaseAsync(flowId, leaseId, leaseDuration, acquire: false, cancellationToken);
 147
 148    public Task ReleaseLeaseAsync(
 149        string flowId,
 150        string leaseId,
 151        CancellationToken cancellationToken = default)
 152    {
 886153        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 886154        ArgumentException.ThrowIfNullOrWhiteSpace(leaseId);
 886155        cancellationToken.ThrowIfCancellationRequested();
 156
 884157        while (_entries.TryGetValue(flowId, out var current))
 158        {
 874159            if (!string.Equals(current.LeaseId, leaseId, StringComparison.Ordinal))
 160                break;
 161
 856162            if (_entries.TryUpdate(flowId, current with { LeaseId = null, LeaseExpiresAtUtc = null }, current))
 163                break;
 164        }
 165
 884166        return Task.CompletedTask;
 167    }
 168
 169    public Task<FlowLeaseObservation?> ObserveLeaseAsync(string flowId, CancellationToken cancellationToken = default)
 170    {
 295171        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 293172        cancellationToken.ThrowIfCancellationRequested();
 173
 174        // Raw, like every other store: an expired lease is reported as persisted. Whether it has
 175        // lapsed is TryAcquireLeaseAsync's call, on this store's clock.
 293176        return Task.FromResult<FlowLeaseObservation?>(
 293177            _entries.TryGetValue(flowId, out var current) && current.LeaseId is not null
 293178                ? new FlowLeaseObservation(current.LeaseId, current.LeaseExpiresAtUtc)
 293179                : FlowLeaseObservation.Unheld);
 180    }
 181
 182    /// <summary>
 183    /// Breaks every held execution lease — the test harness's crash semantics for a simulated
 184    /// restart. A crashed process goes silent and its leases expire; a simulated restart shares
 185    /// the virtual clock with the "crashed" incarnation, whose parked executions would otherwise
 186    /// keep renewing against this shared store forever and the new incarnation could never take
 187    /// their flows over. Breaking the lease makes the zombie's next renewal fail (its lease loop
 188    /// marks itself lost and stops) and lets the new incarnation acquire immediately.
 189    /// </summary>
 190    internal void ExpireAllLeases()
 191    {
 96192        foreach (var flowId in _entries.Keys)
 193        {
 194            // CAS loop: a zombie renewal can swap the entry between the read and the update, and
 195            // TryUpdate compares against the snapshot — a silently lost break would recreate the
 196            // exact "executing on another live worker" hang this method exists to eliminate.
 197            // Retry until no lease is observed; once cleared, the zombie's next renewal fails
 198            // (its lease id no longer matches) and its loop stops, so this converges.
 18199            while (_entries.TryGetValue(flowId, out var entry)
 18200                   && entry.LeaseId is not null
 18201                   && !_entries.TryUpdate(flowId, entry with { LeaseId = null, LeaseExpiresAtUtc = null }, entry))
 202            {
 203            }
 204        }
 30205    }
 206
 207    public Task<bool> TryDeleteAsync(string flowId, CancellationToken cancellationToken = default)
 208    {
 14209        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 14210        cancellationToken.ThrowIfCancellationRequested();
 12211        return Task.FromResult(_entries.TryRemove(flowId, out _));
 212    }
 213
 214    /// <summary>
 215    /// Removes every entry expired at <paramref name="now"/>, at most once per
 216    /// <see cref="SweepInterval"/>; one sweeper at a time (the interval stamp is claimed by
 217    /// compare-exchange). Removal is conditional on the observed entry, so a concurrent
 218    /// update/create that swapped the entry in between keeps its (unexpired) replacement.
 219    /// </summary>
 220    private void SweepExpired(DateTime now)
 221    {
 3692222        var due = Interlocked.Read(ref _nextSweepTicks);
 3692223        if (now.Ticks < due)
 3102224            return;
 225
 590226        var next = Expiry(now, SweepInterval).Ticks;
 590227        if (Interlocked.CompareExchange(ref _nextSweepTicks, next, due) != due)
 0228            return;
 229
 5220230        foreach (var pair in _entries)
 231        {
 2020232            if (pair.Value.ExpiresAtUtc <= now)
 2000233                _entries.TryRemove(pair);
 234        }
 590235    }
 236
 237    private Task<bool> TryChangeLeaseAsync(
 238        string flowId,
 239        string leaseId,
 240        TimeSpan leaseDuration,
 241        bool acquire,
 242        CancellationToken cancellationToken)
 243    {
 3017244        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 3017245        ArgumentException.ThrowIfNullOrWhiteSpace(leaseId);
 3017246        if (leaseDuration <= TimeSpan.Zero)
 2247            throw new ArgumentOutOfRangeException(nameof(leaseDuration), "Lease duration must be greater than zero.");
 3015248        cancellationToken.ThrowIfCancellationRequested();
 249
 3013250        while (_entries.TryGetValue(flowId, out var current))
 251        {
 3005252            var now = _timeProvider.GetUtcNow().UtcDateTime;
 3005253            if (current.ExpiresAtUtc <= now)
 4254                return Task.FromResult(false);
 255
 3001256            var ownsLease = string.Equals(current.LeaseId, leaseId, StringComparison.Ordinal);
 3001257            if (acquire ? current.LeaseId is not null && current.LeaseExpiresAtUtc > now && !ownsLease : !ownsLease || c
 343258                return Task.FromResult(false);
 259
 2658260            var updated = current with
 2658261            {
 2658262                LeaseId = leaseId,
 2658263                LeaseExpiresAtUtc = now.Add(leaseDuration)
 2658264            };
 2658265            if (_entries.TryUpdate(flowId, updated, current))
 2658266                return Task.FromResult(true);
 267        }
 268
 8269        return Task.FromResult(false);
 270    }
 271
 272    private static Entry CreateEntry(
 273        FlowState state,
 274        DateTime expiresAtUtc,
 275        string? leaseId = null,
 276        DateTime? leaseExpiresAtUtc = null)
 8852277        => new(FlowStateJson.Serialize(state), state.Revision, expiresAtUtc, leaseId, leaseExpiresAtUtc);
 278
 279    private static void ValidateWrite(string flowId, FlowState state, TimeSpan ttl)
 280    {
 8918281        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 8918282        ArgumentNullException.ThrowIfNull(state);
 8918283        if (!string.Equals(state.FlowId, flowId, StringComparison.Ordinal))
 4284            throw new ArgumentException("The flow state id must match the store key.", nameof(state));
 8914285        if (state.SchemaVersion != FlowStateSchema.Current)
 4286            throw new ArgumentException("The flow state must use the current schema version.", nameof(state));
 8910287        if (ttl <= TimeSpan.Zero)
 2288            throw new ArgumentOutOfRangeException(nameof(ttl), "TTL must be greater than zero.");
 8908289    }
 290
 8860291    private sealed record Entry(
 7920292        string StateJson,
 13112293        long Revision,
 18535294        DateTime ExpiresAtUtc,
 18967295        string? LeaseId = null,
 24443296        DateTime? LeaseExpiresAtUtc = null);
 297}
Generated by: ReportGenerator 5.5.11.0
09/22/2026 - 09:41:04
GitHub | reportgenerator.io

Methods/Properties

Expiry(System.DateTime,System.TimeSpan)
.cctor()
.ctor(System.TimeProvider)
TryCreateAsync(System.String,AsyncResponse.FlowState,System.TimeSpan,System.Threading.CancellationToken)
LoadAsync(System.String,System.Threading.CancellationToken)
TryUpdateAsync(System.String,AsyncResponse.FlowState,System.Int64,System.TimeSpan,System.String,System.Threading.CancellationToken)
TryAcquireLeaseAsync(System.String,System.String,System.TimeSpan,System.Threading.CancellationToken)
TryRenewLeaseAsync(System.String,System.String,System.TimeSpan,System.Threading.CancellationToken)
ReleaseLeaseAsync(System.String,System.String,System.Threading.CancellationToken)
ObserveLeaseAsync(System.String,System.Threading.CancellationToken)
ExpireAllLeases()
TryDeleteAsync(System.String,System.Threading.CancellationToken)
SweepExpired(System.DateTime)
TryChangeLeaseAsync(System.String,System.String,System.TimeSpan,System.Boolean,System.Threading.CancellationToken)
CreateEntry(AsyncResponse.FlowState,System.DateTime,System.String,System.Nullable`1<System.DateTime>)
ValidateWrite(System.String,AsyncResponse.FlowState,System.TimeSpan)
.ctor(System.String,System.Int64,System.DateTime,System.String,System.Nullable`1<System.DateTime>)
get_StateJson()
get_Revision()
get_ExpiresAtUtc()
get_LeaseId()
get_LeaseExpiresAtUtc()