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

Information
Class: AsyncResponse.InMemoryFlowStateStore.Entry
Assembly: AsyncResponse.Core
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/AsyncResponse.Core/InMemoryFlowStateStore.cs
Line coverage
100%
Covered lines: 6
Uncovered lines: 0
Coverable lines: 6
Total lines: 204
Line coverage: 100%
Branch coverage
N/A
Covered branches: 0
Total branches: 0
Branch coverage: N/A
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.ctor(...)100%11100%

File(s)

/home/runner/work/AsyncResponse/AsyncResponse/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    private readonly ConcurrentDictionary<string, Entry> _entries = new(StringComparer.Ordinal);
 9
 10    public Task<bool> TryCreateAsync(
 11        string flowId,
 12        FlowState state,
 13        TimeSpan ttl,
 14        CancellationToken cancellationToken = default)
 15    {
 16        ValidateWrite(flowId, state, ttl);
 17        cancellationToken.ThrowIfCancellationRequested();
 18        if (state.Revision != 0)
 19            throw new ArgumentException("A new flow ledger must start at revision zero.", nameof(state));
 20
 21        var now = DateTime.UtcNow;
 22        var created = CreateEntry(state, now.Add(ttl));
 23        while (true)
 24        {
 25            if (_entries.TryAdd(flowId, created))
 26                return Task.FromResult(true);
 27
 28            if (!_entries.TryGetValue(flowId, out var existing))
 29                continue;
 30
 31            if (existing.ExpiresAtUtc > now)
 32                return Task.FromResult(false);
 33
 34            if (_entries.TryUpdate(flowId, created, existing))
 35                return Task.FromResult(true);
 36        }
 37    }
 38
 39    public Task<FlowState?> LoadAsync(string flowId, CancellationToken cancellationToken = default)
 40    {
 41        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 42        cancellationToken.ThrowIfCancellationRequested();
 43
 44        while (_entries.TryGetValue(flowId, out var entry))
 45        {
 46            if (entry.ExpiresAtUtc <= DateTime.UtcNow)
 47            {
 48                _entries.TryRemove(KeyValuePair.Create(flowId, entry));
 49                continue;
 50            }
 51
 52            var state = FlowStateJson.Deserialize(entry.StateJson);
 53            return Task.FromResult(
 54                state is not null
 55                && FlowStateSchema.IsReadable(state.SchemaVersion)
 56                && state.Revision == entry.Revision
 57                && string.Equals(state.FlowId, flowId, StringComparison.Ordinal)
 58                    ? state
 59                    : null);
 60        }
 61
 62        return Task.FromResult<FlowState?>(null);
 63    }
 64
 65    public Task<bool> TryUpdateAsync(
 66        string flowId,
 67        FlowState state,
 68        long expectedRevision,
 69        TimeSpan ttl,
 70        string? leaseId = null,
 71        CancellationToken cancellationToken = default)
 72    {
 73        ValidateWrite(flowId, state, ttl);
 74        cancellationToken.ThrowIfCancellationRequested();
 75        if (expectedRevision < 0)
 76            throw new ArgumentOutOfRangeException(nameof(expectedRevision), "Expected revision cannot be negative.");
 77        if (state.Revision != checked(expectedRevision + 1))
 78            throw new ArgumentException("The new flow-state revision must increment the expected revision by one.", name
 79
 80        while (_entries.TryGetValue(flowId, out var current))
 81        {
 82            var now = DateTime.UtcNow;
 83            if (current.ExpiresAtUtc <= now || current.Revision != expectedRevision)
 84                return Task.FromResult(false);
 85            if (leaseId is not null
 86                && (!string.Equals(current.LeaseId, leaseId, StringComparison.Ordinal)
 87                    || current.LeaseExpiresAtUtc <= now))
 88                return Task.FromResult(false);
 89
 90            var updated = CreateEntry(
 91                state,
 92                now.Add(ttl),
 93                current.LeaseId,
 94                current.LeaseExpiresAtUtc);
 95            if (_entries.TryUpdate(flowId, updated, current))
 96                return Task.FromResult(true);
 97        }
 98
 99        return Task.FromResult(false);
 100    }
 101
 102    public Task<bool> TryAcquireLeaseAsync(
 103        string flowId,
 104        string leaseId,
 105        TimeSpan leaseDuration,
 106        CancellationToken cancellationToken = default)
 107        => TryChangeLeaseAsync(flowId, leaseId, leaseDuration, acquire: true, cancellationToken);
 108
 109    public Task<bool> TryRenewLeaseAsync(
 110        string flowId,
 111        string leaseId,
 112        TimeSpan leaseDuration,
 113        CancellationToken cancellationToken = default)
 114        => TryChangeLeaseAsync(flowId, leaseId, leaseDuration, acquire: false, cancellationToken);
 115
 116    public Task ReleaseLeaseAsync(
 117        string flowId,
 118        string leaseId,
 119        CancellationToken cancellationToken = default)
 120    {
 121        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 122        ArgumentException.ThrowIfNullOrWhiteSpace(leaseId);
 123        cancellationToken.ThrowIfCancellationRequested();
 124
 125        while (_entries.TryGetValue(flowId, out var current))
 126        {
 127            if (!string.Equals(current.LeaseId, leaseId, StringComparison.Ordinal))
 128                break;
 129
 130            if (_entries.TryUpdate(flowId, current with { LeaseId = null, LeaseExpiresAtUtc = null }, current))
 131                break;
 132        }
 133
 134        return Task.CompletedTask;
 135    }
 136
 137    public Task<bool> TryDeleteAsync(string flowId, CancellationToken cancellationToken = default)
 138    {
 139        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 140        cancellationToken.ThrowIfCancellationRequested();
 141        return Task.FromResult(_entries.TryRemove(flowId, out _));
 142    }
 143
 144    private Task<bool> TryChangeLeaseAsync(
 145        string flowId,
 146        string leaseId,
 147        TimeSpan leaseDuration,
 148        bool acquire,
 149        CancellationToken cancellationToken)
 150    {
 151        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 152        ArgumentException.ThrowIfNullOrWhiteSpace(leaseId);
 153        if (leaseDuration <= TimeSpan.Zero)
 154            throw new ArgumentOutOfRangeException(nameof(leaseDuration), "Lease duration must be greater than zero.");
 155        cancellationToken.ThrowIfCancellationRequested();
 156
 157        while (_entries.TryGetValue(flowId, out var current))
 158        {
 159            var now = DateTime.UtcNow;
 160            if (current.ExpiresAtUtc <= now)
 161                return Task.FromResult(false);
 162
 163            var ownsLease = string.Equals(current.LeaseId, leaseId, StringComparison.Ordinal);
 164            if (acquire ? current.LeaseId is not null && current.LeaseExpiresAtUtc > now && !ownsLease : !ownsLease || c
 165                return Task.FromResult(false);
 166
 167            var updated = current with
 168            {
 169                LeaseId = leaseId,
 170                LeaseExpiresAtUtc = now.Add(leaseDuration)
 171            };
 172            if (_entries.TryUpdate(flowId, updated, current))
 173                return Task.FromResult(true);
 174        }
 175
 176        return Task.FromResult(false);
 177    }
 178
 179    private static Entry CreateEntry(
 180        FlowState state,
 181        DateTime expiresAtUtc,
 182        string? leaseId = null,
 183        DateTime? leaseExpiresAtUtc = null)
 184        => new(FlowStateJson.Serialize(state), state.Revision, expiresAtUtc, leaseId, leaseExpiresAtUtc);
 185
 186    private static void ValidateWrite(string flowId, FlowState state, TimeSpan ttl)
 187    {
 188        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 189        ArgumentNullException.ThrowIfNull(state);
 190        if (!string.Equals(state.FlowId, flowId, StringComparison.Ordinal))
 191            throw new ArgumentException("The flow state id must match the store key.", nameof(state));
 192        if (state.SchemaVersion != FlowStateSchema.Current)
 193            throw new ArgumentException("The flow state must use the current schema version.", nameof(state));
 194        if (ttl <= TimeSpan.Zero)
 195            throw new ArgumentOutOfRangeException(nameof(ttl), "TTL must be greater than zero.");
 196    }
 197
 3198    private sealed record Entry(
 3199        string StateJson,
 3200        long Revision,
 3201        DateTime ExpiresAtUtc,
 3202        string? LeaseId = null,
 3203        DateTime? LeaseExpiresAtUtc = null);
 204}
Generated by: ReportGenerator 5.5.10.0
08/03/2026 - 08:59:44
GitHub | reportgenerator.io

Methods/Properties

.ctor(string, long, System.DateTime, string, System.Nullable<System.DateTime>)