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

Information
Class: AsyncResponse.InMemoryRecoveryStateStore
Assembly: AsyncResponse.Core
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/AsyncResponse.Core/InMemoryRecoveryStateStore.cs
Line coverage
96%
Covered lines: 149
Uncovered lines: 5
Coverable lines: 154
Total lines: 334
Line coverage: 96.7%
Branch coverage
93%
Covered branches: 118
Total branches: 126
Branch coverage: 93.6%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.ctor(...)100%11100%
.ctor()100%11100%
SaveAsync(...)100%1414100%
GetAllAsync(...)100%88100%
TryDeleteAsync(...)83.33%1818100%
TryRemove(...)100%11100%
.ctor(...)100%11100%
get_IsEmpty()100%22100%
get_SingleEntry()100%11100%
get_ManyEntries()100%11100%
Single(...)100%11100%
Upsert(...)100%1010100%
PruneExpired(...)95%202095.45%
Remove(...)94.44%181891.3%
ReadableStates()87.5%171687.5%
Equals(...)100%22100%
ScanAsync()94.44%1818100%

File(s)

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

#LineLine coverage
 1using System.Collections.Concurrent;
 2using System.Runtime.CompilerServices;
 3
 4namespace AsyncResponse;
 5
 6/// <summary>
 7/// Process-local recovery state store. Useful for the default no-infrastructure setup, tests,
 8/// and single-process apps. It is intentionally not durable: entries disappear when the process
 9/// exits.
 10/// </summary>
 11internal sealed class InMemoryRecoveryStateStore : IRecoveryStateStore, IRecoveryStateScanner
 12{
 213    private sealed record Entry(RecoveryState State, DateTime ExpiresAtUtc);
 14
 215    private readonly ConcurrentDictionary<string, EntryBucket> _entries = new(StringComparer.Ordinal);
 16
 17    /// <inheritdoc />
 18    public Task SaveAsync(
 19        string correlationId,
 20        RecoveryState state,
 21        TimeSpan ttl,
 22        CancellationToken cancellationToken = default)
 23    {
 224        ArgumentException.ThrowIfNullOrWhiteSpace(correlationId);
 225        ArgumentNullException.ThrowIfNull(state);
 226        if (ttl <= TimeSpan.Zero)
 227            throw new ArgumentOutOfRangeException(nameof(ttl), "TTL must be greater than zero.");
 228        if (!string.Equals(state.CorrelationId, correlationId, StringComparison.Ordinal))
 229            throw new ArgumentException("The recovery-state correlation id must match the store key.", nameof(state));
 230        if (state.SchemaVersion != RecoveryStateSchema.Current)
 231            throw new ArgumentException("The recovery state must use the current schema version.", nameof(state));
 32
 233        cancellationToken.ThrowIfCancellationRequested();
 234        if (state.RegistrationId == Guid.Empty)
 235            state.RegistrationId = Guid.NewGuid();
 36
 237        var nowUtc = DateTime.UtcNow;
 238        var entry = new Entry(state, nowUtc.Add(ttl));
 39        while (true)
 40        {
 241            if (!_entries.TryGetValue(correlationId, out var bucket))
 42            {
 243                if (_entries.TryAdd(correlationId, EntryBucket.Single(entry)))
 244                    return Task.CompletedTask;
 45
 46                continue;
 47            }
 48
 249            var next = bucket.PruneExpired(nowUtc).Upsert(entry);
 250            if (_entries.TryUpdate(correlationId, next, bucket))
 251                return Task.CompletedTask;
 52        }
 53    }
 54
 55    /// <inheritdoc />
 56    public Task<IReadOnlyList<RecoveryState>> GetAllAsync(string correlationId, CancellationToken cancellationToken = de
 57    {
 258        ArgumentException.ThrowIfNullOrWhiteSpace(correlationId);
 259        cancellationToken.ThrowIfCancellationRequested();
 60
 261        while (_entries.TryGetValue(correlationId, out var bucket))
 62        {
 263            var pruned = bucket.PruneExpired(DateTime.UtcNow);
 264            if (pruned.IsEmpty)
 65            {
 266                TryRemove(correlationId, bucket);
 267                return Task.FromResult<IReadOnlyList<RecoveryState>>([]);
 68            }
 69
 270            if (!pruned.Equals(bucket) && !_entries.TryUpdate(correlationId, pruned, bucket))
 71                continue;
 72
 273            return Task.FromResult(pruned.ReadableStates());
 74        }
 75
 276        return Task.FromResult<IReadOnlyList<RecoveryState>>([]);
 77    }
 78
 79    /// <inheritdoc />
 80    public Task<bool> TryDeleteAsync(string correlationId, Guid registrationId, CancellationToken cancellationToken = de
 81    {
 282        ArgumentException.ThrowIfNullOrWhiteSpace(correlationId);
 283        if (registrationId == Guid.Empty)
 284            throw new ArgumentException("Registration id cannot be empty.", nameof(registrationId));
 285        cancellationToken.ThrowIfCancellationRequested();
 86
 287        while (_entries.TryGetValue(correlationId, out var bucket))
 88        {
 289            var pruned = bucket.PruneExpired(DateTime.UtcNow);
 290            if (pruned.IsEmpty)
 91            {
 292                TryRemove(correlationId, bucket);
 293                return Task.FromResult(false);
 94            }
 95
 296            var next = pruned.Remove(registrationId, out var removed);
 297            if (!removed)
 98            {
 299                if (!pruned.Equals(bucket) && !_entries.TryUpdate(correlationId, pruned, bucket))
 100                    continue;
 101
 2102                return Task.FromResult(false);
 103            }
 104
 2105            if (next.IsEmpty)
 106            {
 2107                if (TryRemove(correlationId, bucket))
 2108                    return Task.FromResult(true);
 109            }
 2110            else if (_entries.TryUpdate(correlationId, next, bucket))
 111            {
 2112                return Task.FromResult(true);
 113            }
 114        }
 115
 2116        return Task.FromResult(false);
 117    }
 118
 119    private bool TryRemove(string correlationId, EntryBucket bucket)
 2120        => ((ICollection<KeyValuePair<string, EntryBucket>>)_entries)
 2121            .Remove(new KeyValuePair<string, EntryBucket>(correlationId, bucket));
 122
 123    // Deliberately not a flat ConcurrentDictionary<(correlationId, registrationId), Entry>: the
 124    // hot-path lookup is GetAllAsync(correlationId) — every lost-subscriber dispatch — which needs
 125    // all of one correlation id's registrations in O(1)+small-array, and TTL pruning is per-bucket.
 126    // A flat tuple key would make both O(total entries). The reference-identity Equals below is
 127    // what lets a prune+mutate publish atomically via TryUpdate's compare operand.
 128    private readonly struct EntryBucket : IEquatable<EntryBucket>
 129    {
 130        private readonly Entry? _single;
 131        private readonly Entry[]? _many;
 132
 133        private EntryBucket(Entry? single, Entry[]? many)
 134        {
 2135            _single = single;
 2136            _many = many;
 2137        }
 138
 2139        public bool IsEmpty => _single is null && _many is null;
 2140        public Entry? SingleEntry => _single;
 2141        public Entry[]? ManyEntries => _many;
 142
 2143        public static EntryBucket Single(Entry entry) => new(entry, null);
 144
 145        public EntryBucket Upsert(Entry entry)
 146        {
 2147            if (_single is null)
 148            {
 2149                if (_many is null)
 2150                    return Single(entry);
 151
 2152                for (var i = 0; i < _many.Length; i++)
 153                {
 2154                    if (_many[i].State.RegistrationId != entry.State.RegistrationId)
 155                        continue;
 156
 2157                    var replaced = (Entry[])_many.Clone();
 2158                    replaced[i] = entry;
 2159                    return new EntryBucket(null, replaced);
 160                }
 161
 2162                var appended = new Entry[_many.Length + 1];
 2163                Array.Copy(_many, appended, _many.Length);
 2164                appended[^1] = entry;
 2165                return new EntryBucket(null, appended);
 166            }
 167
 2168            if (_single.State.RegistrationId == entry.State.RegistrationId)
 2169                return Single(entry);
 170
 2171            return new EntryBucket(null, [_single, entry]);
 172        }
 173
 174        public EntryBucket PruneExpired(DateTime nowUtc)
 175        {
 2176            if (_single is not null)
 2177                return _single.ExpiresAtUtc <= nowUtc ? default : this;
 178
 2179            if (_many is null)
 0180                return this;
 181
 2182            var liveCount = 0;
 2183            Entry? lastLive = null;
 2184            foreach (var entry in _many)
 185            {
 2186                if (entry.ExpiresAtUtc <= nowUtc)
 187                    continue;
 188
 2189                liveCount++;
 2190                lastLive = entry;
 191            }
 192
 2193            if (liveCount == _many.Length)
 2194                return this;
 2195            if (liveCount == 0)
 2196                return default;
 2197            if (liveCount == 1)
 2198                return Single(lastLive!);
 199
 2200            var live = new Entry[liveCount];
 2201            var index = 0;
 2202            foreach (var entry in _many)
 203            {
 2204                if (entry.ExpiresAtUtc > nowUtc)
 2205                    live[index++] = entry;
 206            }
 207
 2208            return new EntryBucket(null, live);
 209        }
 210
 211        public EntryBucket Remove(Guid registrationId, out bool removed)
 212        {
 2213            if (_single is not null)
 214            {
 2215                removed = _single.State.RegistrationId == registrationId;
 2216                return removed ? default : this;
 217            }
 218
 2219            if (_many is null)
 220            {
 0221                removed = false;
 0222                return this;
 223            }
 224
 2225            var removeIndex = -1;
 2226            for (var i = 0; i < _many.Length; i++)
 227            {
 2228                if (_many[i].State.RegistrationId == registrationId)
 229                {
 2230                    removeIndex = i;
 2231                    break;
 232                }
 233            }
 234
 2235            if (removeIndex < 0)
 236            {
 2237                removed = false;
 2238                return this;
 239            }
 240
 2241            removed = true;
 2242            if (_many.Length == 2)
 2243                return Single(_many[removeIndex == 0 ? 1 : 0]);
 244
 2245            var remaining = new Entry[_many.Length - 1];
 2246            if (removeIndex > 0)
 2247                Array.Copy(_many, 0, remaining, 0, removeIndex);
 2248            if (removeIndex < _many.Length - 1)
 2249                Array.Copy(_many, removeIndex + 1, remaining, removeIndex, _many.Length - removeIndex - 1);
 250
 2251            return new EntryBucket(null, remaining);
 252        }
 253
 254        public IReadOnlyList<RecoveryState> ReadableStates()
 255        {
 2256            if (_single is not null)
 2257                return RecoveryStateSchema.IsReadable(_single.State.SchemaVersion) ? [_single.State] : [];
 258
 2259            if (_many is null)
 0260                return [];
 261
 2262            var readableCount = 0;
 2263            foreach (var entry in _many)
 264            {
 2265                if (RecoveryStateSchema.IsReadable(entry.State.SchemaVersion))
 2266                    readableCount++;
 267            }
 268
 2269            if (readableCount == 0)
 0270                return [];
 271
 2272            var states = new RecoveryState[readableCount];
 2273            var index = 0;
 2274            foreach (var entry in _many)
 275            {
 2276                if (RecoveryStateSchema.IsReadable(entry.State.SchemaVersion))
 2277                    states[index++] = entry.State;
 278            }
 279
 2280            return states;
 281        }
 282
 283        public bool Equals(EntryBucket other)
 2284            => ReferenceEquals(_single, other._single) && ReferenceEquals(_many, other._many);
 285
 286        [System.Diagnostics.CodeAnalysis.ExcludeFromCodeCoverage]
 287        public override bool Equals(object? obj)
 288            => obj is EntryBucket other && Equals(other);
 289
 290        [System.Diagnostics.CodeAnalysis.ExcludeFromCodeCoverage]
 291        public override int GetHashCode()
 292            => HashCode.Combine(
 293                _single is null ? 0 : RuntimeHelpers.GetHashCode(_single),
 294                _many is null ? 0 : RuntimeHelpers.GetHashCode(_many));
 295    }
 296
 297    /// <inheritdoc />
 298    public async IAsyncEnumerable<RecoveryState> ScanAsync([EnumeratorCancellation] CancellationToken cancellationToken 
 299    {
 2300        await Task.CompletedTask.ConfigureAwait(false); // process-local store: no async I/O to await
 301
 2302        var nowUtc = DateTime.UtcNow;
 2303        foreach (var (correlationId, bucket) in _entries)
 304        {
 2305            cancellationToken.ThrowIfCancellationRequested();
 306
 2307            var pruned = bucket.PruneExpired(nowUtc);
 2308            if (pruned.IsEmpty)
 309            {
 2310                TryRemove(correlationId, bucket);
 2311                continue;
 312            }
 313
 2314            if (!pruned.Equals(bucket))
 2315                _entries.TryUpdate(correlationId, pruned, bucket);
 316
 2317            if (pruned.SingleEntry is { } single)
 318            {
 2319                if (RecoveryStateSchema.IsReadable(single.State.SchemaVersion))
 2320                    yield return single.State;
 321                continue;
 322            }
 323
 2324            if (pruned.ManyEntries is null)
 325                continue;
 326
 2327            foreach (var entry in pruned.ManyEntries)
 328            {
 2329                if (RecoveryStateSchema.IsReadable(entry.State.SchemaVersion))
 2330                    yield return entry.State;
 331            }
 332        }
 2333    }
 334}
Generated by: ReportGenerator 5.5.10.0
08/03/2026 - 08:59:44
GitHub | reportgenerator.io

Methods/Properties

.ctor(AsyncResponse.RecoveryState, System.DateTime)
.ctor()
SaveAsync(string, AsyncResponse.RecoveryState, System.TimeSpan, System.Threading.CancellationToken)
GetAllAsync(string, System.Threading.CancellationToken)
TryDeleteAsync(string, System.Guid, System.Threading.CancellationToken)
TryRemove(string, AsyncResponse.InMemoryRecoveryStateStore.EntryBucket)
.ctor(AsyncResponse.InMemoryRecoveryStateStore.Entry, AsyncResponse.InMemoryRecoveryStateStore.Entry[])
get_IsEmpty()
get_SingleEntry()
get_ManyEntries()
Single(AsyncResponse.InMemoryRecoveryStateStore.Entry)
Upsert(AsyncResponse.InMemoryRecoveryStateStore.Entry)
PruneExpired(System.DateTime)
Remove(System.Guid, out bool)
ReadableStates()
Equals(AsyncResponse.InMemoryRecoveryStateStore.EntryBucket)
ScanAsync()