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

Information
Class: AsyncResponse.InMemoryRecoveryStateStore
Assembly: AsyncResponse.Core
File(s): /_/src/AsyncResponse.Core/InMemoryRecoveryStateStore.cs
Line coverage
96%
Covered lines: 151
Uncovered lines: 5
Coverable lines: 156
Total lines: 339
Line coverage: 96.7%
Branch coverage
91%
Covered branches: 115
Total branches: 126
Branch coverage: 91.2%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
get_State()100%11100%
.ctor(...)100%22100%
SaveAsync(...)85.71%1414100%
GetAllAsync(...)87.5%88100%
TryDeleteAsync(...)77.77%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()100%1616100%

File(s)

/_/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{
 1493413    private sealed record Entry(RecoveryState State, DateTime ExpiresAtUtc);
 14
 127615    private readonly ConcurrentDictionary<string, EntryBucket> _entries = new(StringComparer.Ordinal);
 16    private readonly TimeProvider _timeProvider;
 17
 18    /// <summary>Creates the store; expiry stamps come from the engine's clock.</summary>
 127619    public InMemoryRecoveryStateStore(TimeProvider? timeProvider = null)
 127620        => _timeProvider = timeProvider ?? TimeProvider.System;
 21
 22    /// <inheritdoc />
 23    public Task SaveAsync(
 24        string correlationId,
 25        RecoveryState state,
 26        TimeSpan ttl,
 27        CancellationToken cancellationToken = default)
 28    {
 402329        ArgumentException.ThrowIfNullOrWhiteSpace(correlationId);
 402130        ArgumentNullException.ThrowIfNull(state);
 401931        if (ttl <= TimeSpan.Zero)
 232            throw new ArgumentOutOfRangeException(nameof(ttl), "TTL must be greater than zero.");
 401733        if (!string.Equals(state.CorrelationId, correlationId, StringComparison.Ordinal))
 434            throw new ArgumentException("The recovery-state correlation id must match the store key.", nameof(state));
 401335        if (state.SchemaVersion != RecoveryStateSchema.Current)
 636            throw new ArgumentException("The recovery state must use the current schema version.", nameof(state));
 37
 400738        cancellationToken.ThrowIfCancellationRequested();
 400539        if (state.RegistrationId == Guid.Empty)
 2840            state.RegistrationId = Guid.NewGuid();
 41
 400542        var nowUtc = _timeProvider.GetUtcNow().UtcDateTime;
 400543        var entry = new Entry(state, nowUtc.Add(ttl));
 44        while (true)
 45        {
 400546            if (!_entries.TryGetValue(correlationId, out var bucket))
 47            {
 383848                if (_entries.TryAdd(correlationId, EntryBucket.Single(entry)))
 383849                    return Task.CompletedTask;
 50
 51                continue;
 52            }
 53
 16754            var next = bucket.PruneExpired(nowUtc).Upsert(entry);
 16755            if (_entries.TryUpdate(correlationId, next, bucket))
 16756                return Task.CompletedTask;
 57        }
 58    }
 59
 60    /// <inheritdoc />
 61    public Task<IReadOnlyList<RecoveryState>> GetAllAsync(string correlationId, CancellationToken cancellationToken = de
 62    {
 32063        ArgumentException.ThrowIfNullOrWhiteSpace(correlationId);
 31864        cancellationToken.ThrowIfCancellationRequested();
 65
 31666        while (_entries.TryGetValue(correlationId, out var bucket))
 67        {
 24068            var pruned = bucket.PruneExpired(_timeProvider.GetUtcNow().UtcDateTime);
 24069            if (pruned.IsEmpty)
 70            {
 671                TryRemove(correlationId, bucket);
 672                return Task.FromResult<IReadOnlyList<RecoveryState>>([]);
 73            }
 74
 23475            if (!pruned.Equals(bucket) && !_entries.TryUpdate(correlationId, pruned, bucket))
 76                continue;
 77
 23478            return Task.FromResult(pruned.ReadableStates());
 79        }
 80
 7681        return Task.FromResult<IReadOnlyList<RecoveryState>>([]);
 82    }
 83
 84    /// <inheritdoc />
 85    public Task<bool> TryDeleteAsync(string correlationId, Guid registrationId, CancellationToken cancellationToken = de
 86    {
 381187        ArgumentException.ThrowIfNullOrWhiteSpace(correlationId);
 381188        if (registrationId == Guid.Empty)
 289            throw new ArgumentException("Registration id cannot be empty.", nameof(registrationId));
 380990        cancellationToken.ThrowIfCancellationRequested();
 91
 380792        while (_entries.TryGetValue(correlationId, out var bucket))
 93        {
 380194            var pruned = bucket.PruneExpired(_timeProvider.GetUtcNow().UtcDateTime);
 380195            if (pruned.IsEmpty)
 96            {
 497                TryRemove(correlationId, bucket);
 498                return Task.FromResult(false);
 99            }
 100
 3797101            var next = pruned.Remove(registrationId, out var removed);
 3797102            if (!removed)
 103            {
 4104                if (!pruned.Equals(bucket) && !_entries.TryUpdate(correlationId, pruned, bucket))
 105                    continue;
 106
 4107                return Task.FromResult(false);
 108            }
 109
 3793110            if (next.IsEmpty)
 111            {
 3720112                if (TryRemove(correlationId, bucket))
 3720113                    return Task.FromResult(true);
 114            }
 73115            else if (_entries.TryUpdate(correlationId, next, bucket))
 116            {
 73117                return Task.FromResult(true);
 118            }
 119        }
 120
 6121        return Task.FromResult(false);
 122    }
 123
 124    private bool TryRemove(string correlationId, EntryBucket bucket)
 3736125        => ((ICollection<KeyValuePair<string, EntryBucket>>)_entries)
 3736126            .Remove(new KeyValuePair<string, EntryBucket>(correlationId, bucket));
 127
 128    // Deliberately not a flat ConcurrentDictionary<(correlationId, registrationId), Entry>: the
 129    // hot-path lookup is GetAllAsync(correlationId) — every lost-subscriber dispatch — which needs
 130    // all of one correlation id's registrations in O(1)+small-array, and TTL pruning is per-bucket.
 131    // A flat tuple key would make both O(total entries). The reference-identity Equals below is
 132    // what lets a prune+mutate publish atomically via TryUpdate's compare operand.
 133    private readonly struct EntryBucket : IEquatable<EntryBucket>
 134    {
 135        private readonly Entry? _single;
 136        private readonly Entry[]? _many;
 137
 138        private EntryBucket(Entry? single, Entry[]? many)
 139        {
 4082140            _single = single;
 4082141            _many = many;
 4082142        }
 143
 7887144        public bool IsEmpty => _single is null && _many is null;
 47145        public Entry? SingleEntry => _single;
 12146        public Entry[]? ManyEntries => _many;
 147
 3889148        public static EntryBucket Single(Entry entry) => new(entry, null);
 149
 150        public EntryBucket Upsert(Entry entry)
 151        {
 167152            if (_single is null)
 153            {
 70154                if (_many is null)
 2155                    return Single(entry);
 156
 1012157                for (var i = 0; i < _many.Length; i++)
 158                {
 440159                    if (_many[i].State.RegistrationId != entry.State.RegistrationId)
 160                        continue;
 161
 2162                    var replaced = (Entry[])_many.Clone();
 2163                    replaced[i] = entry;
 2164                    return new EntryBucket(null, replaced);
 165                }
 166
 66167                var appended = new Entry[_many.Length + 1];
 66168                Array.Copy(_many, appended, _many.Length);
 66169                appended[^1] = entry;
 66170                return new EntryBucket(null, appended);
 171            }
 172
 97173            if (_single.State.RegistrationId == entry.State.RegistrationId)
 2174                return Single(entry);
 175
 95176            return new EntryBucket(null, [_single, entry]);
 177        }
 178
 179        public EntryBucket PruneExpired(DateTime nowUtc)
 180        {
 4261181            if (_single is not null)
 4016182                return _single.ExpiresAtUtc <= nowUtc ? default : this;
 183
 245184            if (_many is null)
 0185                return this;
 186
 245187            var liveCount = 0;
 245188            Entry? lastLive = null;
 2266189            foreach (var entry in _many)
 190            {
 888191                if (entry.ExpiresAtUtc <= nowUtc)
 192                    continue;
 193
 876194                liveCount++;
 876195                lastLive = entry;
 196            }
 197
 245198            if (liveCount == _many.Length)
 237199                return this;
 8200            if (liveCount == 0)
 4201                return default;
 4202            if (liveCount == 1)
 2203                return Single(lastLive!);
 204
 2205            var live = new Entry[liveCount];
 2206            var index = 0;
 16207            foreach (var entry in _many)
 208            {
 6209                if (entry.ExpiresAtUtc > nowUtc)
 4210                    live[index++] = entry;
 211            }
 212
 2213            return new EntryBucket(null, live);
 214        }
 215
 216        public EntryBucket Remove(Guid registrationId, out bool removed)
 217        {
 3797218            if (_single is not null)
 219            {
 3722220                removed = _single.State.RegistrationId == registrationId;
 3722221                return removed ? default : this;
 222            }
 223
 75224            if (_many is null)
 225            {
 0226                removed = false;
 0227                return this;
 228            }
 229
 75230            var removeIndex = -1;
 238231            for (var i = 0; i < _many.Length; i++)
 232            {
 117233                if (_many[i].State.RegistrationId == registrationId)
 234                {
 73235                    removeIndex = i;
 73236                    break;
 237                }
 238            }
 239
 75240            if (removeIndex < 0)
 241            {
 2242                removed = false;
 2243                return this;
 244            }
 245
 73246            removed = true;
 73247            if (_many.Length == 2)
 45248                return Single(_many[removeIndex == 0 ? 1 : 0]);
 249
 28250            var remaining = new Entry[_many.Length - 1];
 28251            if (removeIndex > 0)
 20252                Array.Copy(_many, 0, remaining, 0, removeIndex);
 28253            if (removeIndex < _many.Length - 1)
 20254                Array.Copy(_many, removeIndex + 1, remaining, removeIndex, _many.Length - removeIndex - 1);
 255
 28256            return new EntryBucket(null, remaining);
 257        }
 258
 259        public IReadOnlyList<RecoveryState> ReadableStates()
 260        {
 234261            if (_single is not null)
 144262                return RecoveryStateSchema.IsReadable(_single.State.SchemaVersion) ? [_single.State] : [];
 263
 90264            if (_many is null)
 0265                return [];
 266
 90267            var readableCount = 0;
 660268            foreach (var entry in _many)
 269            {
 240270                if (RecoveryStateSchema.IsReadable(entry.State.SchemaVersion))
 238271                    readableCount++;
 272            }
 273
 90274            if (readableCount == 0)
 0275                return [];
 276
 90277            var states = new RecoveryState[readableCount];
 90278            var index = 0;
 660279            foreach (var entry in _many)
 280            {
 240281                if (RecoveryStateSchema.IsReadable(entry.State.SchemaVersion))
 238282                    states[index++] = entry.State;
 283            }
 284
 90285            return states;
 286        }
 287
 288        public bool Equals(EntryBucket other)
 4265289            => ReferenceEquals(_single, other._single) && ReferenceEquals(_many, other._many);
 290
 291        [System.Diagnostics.CodeAnalysis.ExcludeFromCodeCoverage]
 292        public override bool Equals(object? obj)
 293            => obj is EntryBucket other && Equals(other);
 294
 295        [System.Diagnostics.CodeAnalysis.ExcludeFromCodeCoverage]
 296        public override int GetHashCode()
 297            => HashCode.Combine(
 298                _single is null ? 0 : RuntimeHelpers.GetHashCode(_single),
 299                _many is null ? 0 : RuntimeHelpers.GetHashCode(_many));
 300    }
 301
 302    /// <inheritdoc />
 303    public async IAsyncEnumerable<RecoveryState> ScanAsync([EnumeratorCancellation] CancellationToken cancellationToken 
 304    {
 1277305        await Task.CompletedTask.ConfigureAwait(false); // process-local store: no async I/O to await
 306
 1277307        var nowUtc = _timeProvider.GetUtcNow().UtcDateTime;
 2662308        foreach (var (correlationId, bucket) in _entries)
 309        {
 55310            cancellationToken.ThrowIfCancellationRequested();
 311
 53312            var pruned = bucket.PruneExpired(nowUtc);
 53313            if (pruned.IsEmpty)
 314            {
 6315                TryRemove(correlationId, bucket);
 6316                continue;
 317            }
 318
 47319            if (!pruned.Equals(bucket))
 2320                _entries.TryUpdate(correlationId, pruned, bucket);
 321
 47322            if (pruned.SingleEntry is { } single)
 323            {
 41324                if (RecoveryStateSchema.IsReadable(single.State.SchemaVersion))
 39325                    yield return single.State;
 326                continue;
 327            }
 328
 6329            if (pruned.ManyEntries is null)
 330                continue;
 331
 36332            foreach (var entry in pruned.ManyEntries)
 333            {
 12334                if (RecoveryStateSchema.IsReadable(entry.State.SchemaVersion))
 10335                    yield return entry.State;
 336            }
 337        }
 1275338    }
 339}
Generated by: ReportGenerator 5.5.11.0
09/22/2026 - 09:41:04
GitHub | reportgenerator.io

Methods/Properties

get_State()
.ctor(System.TimeProvider)
SaveAsync(System.String,AsyncResponse.RecoveryState,System.TimeSpan,System.Threading.CancellationToken)
GetAllAsync(System.String,System.Threading.CancellationToken)
TryDeleteAsync(System.String,System.Guid,System.Threading.CancellationToken)
TryRemove(System.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,System.Boolean&)
ReadableStates()
Equals(AsyncResponse.InMemoryRecoveryStateStore/EntryBucket)
ScanAsync()