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

Information
Class: AsyncResponse.Channels.NATS.NatsRecoveryStateStore
Assembly: AsyncResponse.Channels.NATS
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/Channels/AsyncResponse.Channels.NATS/NatsRecoveryStateStore.cs
Line coverage
100%
Covered lines: 138
Uncovered lines: 0
Coverable lines: 138
Total lines: 284
Line coverage: 100%
Branch coverage
98%
Covered branches: 77
Total branches: 78
Branch coverage: 98.7%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.ctor(...)100%22100%
SaveAsync()100%2424100%
GetAllAsync()100%88100%
TryDeleteAsync()100%2020100%
ScanAsync()90%1010100%
LoadStoredAsync()100%22100%
SerializeStates(...)100%11100%
IsExpired(...)100%11100%
IsStateReadable(...)100%88100%
TryDeserialize(...)100%11100%
StatesFrom(...)100%44100%
TryDeleteSilentlyAsync()100%11100%

File(s)

/home/runner/work/AsyncResponse/AsyncResponse/src/Channels/AsyncResponse.Channels.NATS/NatsRecoveryStateStore.cs

#LineLine coverage
 1using Microsoft.Extensions.Logging;
 2using Microsoft.Extensions.Options;
 3using System.Runtime.CompilerServices;
 4using System.Text.Json;
 5using System.Text.Json.Serialization;
 6
 7namespace AsyncResponse.Channels.NATS;
 8
 9/// <summary>
 10/// NATS JetStream Key-Value implementation of <see cref="IRecoveryStateStore"/> and
 11/// <see cref="IRecoveryStateScanner"/>.
 12/// <para>
 13/// NATS KV applies a single <c>MaxAge</c> per bucket rather than a TTL per key, so each entry also
 14/// carries an absolute <see cref="StoredRecoveryState.ExpiresAtUtc"/>: reads and scans treat an entry
 15/// past its expiry as absent (and delete it best-effort), giving precise per-correlation expiry while
 16/// the bucket's <c>MaxAge</c> acts as a garbage-collection ceiling for orphans.
 17/// </para>
 18/// </summary>
 19internal sealed class NatsRecoveryStateStore : IRecoveryStateStore, IRecoveryStateScanner
 20{
 21    private readonly INatsKvStore _store;
 22    private readonly ILogger<NatsRecoveryStateStore> _logger;
 23    private readonly TimeProvider _timeProvider;
 24
 25    /// <summary>Creates a NATS JetStream Key-Value recovery state store.</summary>
 326    public NatsRecoveryStateStore(
 327        INatsKvStore store,
 328        IOptions<NatsAsyncResponseChannelOptions> options,
 329        ILogger<NatsRecoveryStateStore> logger,
 330        TimeProvider? timeProvider = null)
 31    {
 332        options.Value.Validate();
 333        _store = store;
 334        _logger = logger;
 335        _timeProvider = timeProvider ?? TimeProvider.System;
 336    }
 37
 38    /// <inheritdoc />
 39    public async Task SaveAsync(
 40        string correlationId,
 41        RecoveryState state,
 42        TimeSpan ttl,
 43        CancellationToken cancellationToken = default)
 44    {
 345        ArgumentException.ThrowIfNullOrWhiteSpace(correlationId);
 346        ArgumentNullException.ThrowIfNull(state);
 347        if (ttl <= TimeSpan.Zero)
 348            throw new ArgumentOutOfRangeException(nameof(ttl), "TTL must be greater than zero.");
 349        if (!string.Equals(state.CorrelationId, correlationId, StringComparison.Ordinal))
 350            throw new ArgumentException("The recovery-state correlation id must match the store key.", nameof(state));
 351        if (state.SchemaVersion != RecoveryStateSchema.Current)
 352            throw new ArgumentException("The recovery state must use the current schema version.", nameof(state));
 53
 354        cancellationToken.ThrowIfCancellationRequested();
 355        if (state.RegistrationId == Guid.Empty)
 356            state.RegistrationId = Guid.NewGuid();
 57
 358        var key = NatsSubjectSchema.RecoveryKey(correlationId);
 59
 60        // Revision-conditioned read-modify-write: two waiters registering the same correlation id
 61        // concurrently must both survive.
 362        for (var attempt = 0; attempt < MaxCasAttempts; attempt++)
 63        {
 364            cancellationToken.ThrowIfCancellationRequested();
 65
 366            var entry = await _store.GetAsync(key, cancellationToken).ConfigureAwait(false);
 367            var stored = entry is { } existing ? TryDeserialize(existing.Value, key) : null;
 368            var states = stored is not null && !IsExpired(stored)
 369                ? StatesFrom(stored)
 370                : [];
 371            states.RemoveAll(existingState => !IsStateReadable(existingState, key, correlationId));
 372            states.RemoveAll(existingState => existingState.RegistrationId == state.RegistrationId);
 373            states.Add(state);
 374            var json = SerializeStates(states, _timeProvider.GetUtcNow() + ttl);
 75
 376            var written = entry is { } current
 377                ? await _store.TryUpdateAsync(key, json, current.Revision, cancellationToken).ConfigureAwait(false)
 378                : await _store.TryCreateAsync(key, json, cancellationToken).ConfigureAwait(false);
 379            if (written)
 380                return;
 81        }
 82
 383        throw new InvalidOperationException(
 384            $"Recovery-state save for correlationId '{correlationId}' could not commit after {MaxCasAttempts} optimistic
 385    }
 86
 87    /// <inheritdoc />
 88    public async Task<IReadOnlyList<RecoveryState>> GetAllAsync(string correlationId, CancellationToken cancellationToke
 89    {
 390        ArgumentException.ThrowIfNullOrWhiteSpace(correlationId);
 391        cancellationToken.ThrowIfCancellationRequested();
 92
 393        var key = NatsSubjectSchema.RecoveryKey(correlationId);
 394        var stored = await LoadStoredAsync(key, cancellationToken).ConfigureAwait(false);
 395        if (stored is null)
 396            return [];
 97
 398        if (IsExpired(stored))
 99        {
 100            // Past its logical expiry but still physically present (bucket MaxAge has not collected it
 101            // yet): treat as gone and remove it best-effort so it never resurfaces.
 3102            await TryDeleteSilentlyAsync(key, cancellationToken).ConfigureAwait(false);
 2103            return [];
 104        }
 105
 3106        var states = StatesFrom(stored);
 3107        for (var i = states.Count - 1; i >= 0; i--)
 108        {
 3109            var state = states[i];
 3110            if (!IsStateReadable(state, key, correlationId))
 111            {
 3112                states.RemoveAt(i);
 113                continue;
 114            }
 115        }
 116
 3117        return states;
 3118    }
 119
 120    /// <inheritdoc />
 121    public async Task<bool> TryDeleteAsync(string correlationId, Guid registrationId, CancellationToken cancellationToke
 122    {
 3123        ArgumentException.ThrowIfNullOrWhiteSpace(correlationId);
 3124        if (registrationId == Guid.Empty)
 3125            throw new ArgumentException("Registration id cannot be empty.", nameof(registrationId));
 3126        cancellationToken.ThrowIfCancellationRequested();
 127
 3128        var key = NatsSubjectSchema.RecoveryKey(correlationId);
 129
 130        // Revision-conditioned removal: deleting one registration must not clobber a registration
 131        // that another writer appended between our read and our write.
 3132        for (var attempt = 0; attempt < MaxCasAttempts; attempt++)
 133        {
 3134            cancellationToken.ThrowIfCancellationRequested();
 135
 3136            var entry = await _store.GetAsync(key, cancellationToken).ConfigureAwait(false);
 3137            if (entry is not { } existing)
 3138                return false;
 139
 3140            var stored = TryDeserialize(existing.Value, key);
 3141            if (stored is null)
 3142                return false;
 143
 3144            if (IsExpired(stored))
 145            {
 3146                await TryDeleteSilentlyAsync(key, cancellationToken).ConfigureAwait(false);
 3147                return false;
 148            }
 149
 3150            var states = StatesFrom(stored);
 3151            states.RemoveAll(state => !IsStateReadable(state, key, correlationId));
 3152            var removed = states.RemoveAll(state => state.RegistrationId == registrationId) > 0;
 3153            if (!removed)
 3154                return false;
 155
 3156            var succeeded = states.Count == 0
 3157                ? await _store.TryDeleteAsync(key, existing.Revision, cancellationToken).ConfigureAwait(false)
 3158                : await _store.TryUpdateAsync(key, SerializeStates(states, stored.ExpiresAtUtc), existing.Revision, canc
 3159            if (succeeded)
 3160                return true;
 161        }
 162
 3163        _logger.LogWarning(
 3164            "Recovery-state delete for correlationId {CorrelationId} registration {RegistrationId} exhausted {Attempts} 
 3165            correlationId, registrationId, MaxCasAttempts);
 2166        return false;
 3167    }
 168
 169    /// <inheritdoc />
 170    public async IAsyncEnumerable<RecoveryState> ScanAsync([EnumeratorCancellation] CancellationToken cancellationToken 
 171    {
 3172        await foreach (var key in _store.GetKeysAsync(cancellationToken).ConfigureAwait(false))
 173        {
 3174            cancellationToken.ThrowIfCancellationRequested();
 175
 3176            var stored = await LoadStoredAsync(key, cancellationToken).ConfigureAwait(false);
 3177            if (stored is null)
 178                continue;
 179
 3180            if (IsExpired(stored))
 181            {
 2182                await TryDeleteSilentlyAsync(key, cancellationToken).ConfigureAwait(false);
 3183                continue;
 184            }
 185
 3186            foreach (var state in StatesFrom(stored))
 187            {
 3188                var correlationId = NatsSubjectSchema.CorrelationIdFromRecoveryKey(key);
 3189                if (!IsStateReadable(state, key, correlationId))
 190                    continue;
 191
 3192                yield return state;
 193            }
 3194        }
 3195    }
 196
 197    private const int MaxCasAttempts = 4;
 198
 199    private async Task<StoredRecoveryState?> LoadStoredAsync(string key, CancellationToken cancellationToken)
 200    {
 3201        var entry = await _store.GetAsync(key, cancellationToken).ConfigureAwait(false);
 3202        return entry is { } existing ? TryDeserialize(existing.Value, key) : null;
 3203    }
 204
 205    private static string SerializeStates(List<RecoveryState> states, DateTimeOffset expiresAtUtc)
 3206        => JsonSerializer.Serialize(new StoredRecoveryState
 3207        {
 3208            States = states,
 3209            ExpiresAtUtc = expiresAtUtc
 3210        }, NatsChannelJsonContext.Default.StoredRecoveryState);
 211
 3212    private bool IsExpired(StoredRecoveryState stored) => stored.ExpiresAtUtc <= _timeProvider.GetUtcNow();
 213
 214    private bool IsStateReadable(RecoveryState? state, string key, string correlationId)
 215    {
 3216        if (state is null || state.RegistrationId == Guid.Empty)
 217        {
 3218            _logger.LogWarning(
 3219                "Recovery state at key {RecoveryKey} has no registration id; rejecting it because it cannot be deleted s
 3220                key);
 3221            return false;
 222        }
 223
 3224        if (!RecoveryStateSchema.IsReadable(state.SchemaVersion))
 225        {
 3226            _logger.LogWarning(
 3227                "Recovery state at key {RecoveryKey} has unsupported schema version {SchemaVersion} (current: {Current})
 3228                key, state.SchemaVersion, RecoveryStateSchema.Current);
 3229            return false;
 230        }
 231
 3232        if (string.Equals(state.CorrelationId, correlationId, StringComparison.Ordinal))
 3233            return true;
 234
 3235        _logger.LogWarning(
 3236            "Recovery state at key {RecoveryKey} has correlationId {StoredCorrelationId}, expected {CorrelationId}; reje
 3237            key, state.CorrelationId, correlationId);
 3238        return false;
 239    }
 240
 241    private StoredRecoveryState? TryDeserialize(string json, string key)
 242    {
 243        try
 244        {
 3245            return JsonSerializer.Deserialize(json, NatsChannelJsonContext.Default.StoredRecoveryState);
 246        }
 3247        catch (JsonException ex)
 248        {
 3249            _logger.LogWarning(ex, "Unreadable recovery state at key {RecoveryKey}; skipping.", key);
 3250            return null;
 251        }
 3252    }
 253
 254    private static List<RecoveryState> StatesFrom(StoredRecoveryState stored)
 3255        => stored.States is { Count: > 0 } ? [.. stored.States] : [];
 256
 257    private async Task TryDeleteSilentlyAsync(string key, CancellationToken cancellationToken)
 258    {
 259        try
 260        {
 2261            await _store.DeleteAsync(key, cancellationToken).ConfigureAwait(false);
 2262        }
 2263        catch (Exception ex)
 264        {
 2265            _logger.LogDebug(ex, "Best-effort delete of expired recovery state at key {RecoveryKey} failed.", key);
 2266        }
 2267    }
 268
 269    /// <summary>The stored envelope: the recovery state plus its absolute expiry, for per-key logical TTL.</summary>
 270    internal sealed class StoredRecoveryState
 271    {
 272        public List<RecoveryState>? States { get; set; }
 273        public DateTimeOffset ExpiresAtUtc { get; set; }
 274    }
 275}
 276
 277/// <summary>
 278/// Source-generated metadata for the package-local KV envelope (trim/AOT-safe; the wire format is
 279/// unchanged — Metadata-mode generation with default options matches the previous reflection-based
 280/// serialization exactly).
 281/// </summary>
 282[JsonSourceGenerationOptions(GenerationMode = JsonSourceGenerationMode.Metadata)]
 283[JsonSerializable(typeof(NatsRecoveryStateStore.StoredRecoveryState))]
 284internal sealed partial class NatsChannelJsonContext : JsonSerializerContext;
Generated by: ReportGenerator 5.5.10.0
08/03/2026 - 08:59:44
GitHub | reportgenerator.io

Methods/Properties

.ctor(AsyncResponse.Channels.NATS.INatsKvStore, Microsoft.Extensions.Options.IOptions<AsyncResponse.Channels.NATS.NatsAsyncResponseChannelOptions>, Microsoft.Extensions.Logging.ILogger<AsyncResponse.Channels.NATS.NatsRecoveryStateStore>, System.TimeProvider)
SaveAsync()
GetAllAsync()
TryDeleteAsync()
ScanAsync()
LoadStoredAsync()
SerializeStates(System.Collections.Generic.List<AsyncResponse.RecoveryState>, System.DateTimeOffset)
IsExpired(AsyncResponse.Channels.NATS.NatsRecoveryStateStore.StoredRecoveryState)
IsStateReadable(AsyncResponse.RecoveryState, string, string)
TryDeserialize(string, string)
StatesFrom(AsyncResponse.Channels.NATS.NatsRecoveryStateStore.StoredRecoveryState)
TryDeleteSilentlyAsync()