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

Information
Class: AsyncResponse.Channels.Redis.RedisRecoveryStateStore
Assembly: AsyncResponse.Channels.Redis
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/Channels/AsyncResponse.Channels.Redis/RedisRecoveryStateStore.cs
Line coverage
100%
Covered lines: 119
Uncovered lines: 0
Coverable lines: 119
Total lines: 222
Line coverage: 100%
Branch coverage
98%
Covered branches: 57
Total branches: 58
Branch coverage: 98.2%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.ctor(...)100%11100%
SaveAsync()100%1616100%
GetAllAsync()100%11100%
TryDeleteAsync()100%1212100%
ScanAsync()92.86%1414100%
LoadStatesAsync()100%22100%
DeserializeStates(...)100%1414100%

File(s)

/home/runner/work/AsyncResponse/AsyncResponse/src/Channels/AsyncResponse.Channels.Redis/RedisRecoveryStateStore.cs

#LineLine coverage
 1using Microsoft.Extensions.Logging;
 2using Microsoft.Extensions.Options;
 3using StackExchange.Redis;
 4using System.Runtime.CompilerServices;
 5using System.Text.Json;
 6
 7namespace AsyncResponse.Channels.Redis;
 8
 9/// <summary>
 10/// Redis-backed implementation of <see cref="IRecoveryStateStore"/>.
 11/// </summary>
 12internal sealed class RedisRecoveryStateStore : IRecoveryStateStore, IRecoveryStateScanner
 13{
 14    private readonly IConnectionMultiplexer _multiplexer;
 15    private readonly IDatabase _database;
 16    private readonly RedisKeySchema _keys;
 17    private readonly ILogger<RedisRecoveryStateStore> _logger;
 18
 19    /// <summary>Creates a Redis-backed recovery state store.</summary>
 320    public RedisRecoveryStateStore(
 321        IConnectionMultiplexer multiplexer,
 322        IOptions<RedisAsyncResponseOptions> options,
 323        ILogger<RedisRecoveryStateStore> logger)
 24    {
 325        _multiplexer = multiplexer;
 326        _database = multiplexer.GetDatabase();
 327        _keys = new RedisKeySchema(options.Value.KeyPrefix);
 328        _logger = logger;
 329    }
 30
 31    /// <inheritdoc />
 32    public async Task SaveAsync(
 33        string correlationId,
 34        RecoveryState state,
 35        TimeSpan ttl,
 36        CancellationToken cancellationToken = default)
 37    {
 338        ArgumentException.ThrowIfNullOrWhiteSpace(correlationId);
 339        ArgumentNullException.ThrowIfNull(state);
 340        if (ttl <= TimeSpan.Zero)
 341            throw new ArgumentOutOfRangeException(nameof(ttl), "TTL must be greater than zero.");
 342        if (!string.Equals(state.CorrelationId, correlationId, StringComparison.Ordinal))
 343            throw new ArgumentException("The recovery-state correlation id must match the store key.", nameof(state));
 344        if (state.SchemaVersion != RecoveryStateSchema.Current)
 345            throw new ArgumentException("The recovery state must use the current schema version.", nameof(state));
 46
 347        cancellationToken.ThrowIfCancellationRequested();
 348        if (state.RegistrationId == Guid.Empty)
 349            state.RegistrationId = Guid.NewGuid();
 50
 351        var recoveryKey = _keys.RecoveryKey(correlationId);
 52
 53        // Optimistic read-modify-write: two waiters registering the same correlation id
 54        // concurrently must both survive, so each write commits only while the stored value is
 55        // still the one we read (transaction condition), retrying on a conflict.
 356        for (var attempt = 0; attempt < MaxCasAttempts; attempt++)
 57        {
 358            cancellationToken.ThrowIfCancellationRequested();
 59
 360            var previous = await _database.StringGetAsync(recoveryKey).ConfigureAwait(false);
 361            var states = previous.IsNullOrEmpty
 362                ? []
 363                : DeserializeStates(previous, recoveryKey, correlationId, logAsError: true);
 364            states.RemoveAll(existing => existing.RegistrationId == state.RegistrationId);
 365            states.Add(state);
 66
 367            var transaction = _database.CreateTransaction();
 368            transaction.AddCondition(previous.IsNull
 369                ? Condition.KeyNotExists(recoveryKey)
 370                : Condition.StringEqual(recoveryKey, previous));
 371            _ = transaction.StringSetAsync(recoveryKey, AsyncResponseJson.Serialize(states), ttl);
 372            if (await transaction.ExecuteAsync().ConfigureAwait(false))
 373                return;
 74        }
 75
 376        throw new InvalidOperationException(
 377            $"Recovery-state save for correlationId '{correlationId}' could not commit after {MaxCasAttempts} optimistic
 378    }
 79
 80    /// <inheritdoc />
 81    public async Task<IReadOnlyList<RecoveryState>> GetAllAsync(string correlationId, CancellationToken cancellationToke
 82    {
 383        ArgumentException.ThrowIfNullOrWhiteSpace(correlationId);
 384        cancellationToken.ThrowIfCancellationRequested();
 85
 386        var recoveryKey = _keys.RecoveryKey(correlationId);
 387        return await LoadStatesAsync(recoveryKey, correlationId).ConfigureAwait(false);
 388    }
 89
 90    /// <inheritdoc />
 91    public async Task<bool> TryDeleteAsync(string correlationId, Guid registrationId, CancellationToken cancellationToke
 92    {
 393        ArgumentException.ThrowIfNullOrWhiteSpace(correlationId);
 394        if (registrationId == Guid.Empty)
 395            throw new ArgumentException("Registration id cannot be empty.", nameof(registrationId));
 396        cancellationToken.ThrowIfCancellationRequested();
 97
 398        var recoveryKey = _keys.RecoveryKey(correlationId);
 99
 100        // Optimistic removal: deleting one registration must not clobber a registration that a
 101        // concurrent writer appended between our read and our write.
 3102        for (var attempt = 0; attempt < MaxCasAttempts; attempt++)
 103        {
 3104            cancellationToken.ThrowIfCancellationRequested();
 105
 3106            var previous = await _database.StringGetAsync(recoveryKey).ConfigureAwait(false);
 3107            if (previous.IsNullOrEmpty)
 3108                return false;
 109
 3110            var states = DeserializeStates(previous, recoveryKey, correlationId, logAsError: true);
 3111            var removed = states.RemoveAll(state => state.RegistrationId == registrationId) > 0;
 3112            if (!removed)
 3113                return false;
 114
 3115            var transaction = _database.CreateTransaction();
 3116            transaction.AddCondition(Condition.StringEqual(recoveryKey, previous));
 3117            if (states.Count == 0)
 3118                _ = transaction.KeyDeleteAsync(recoveryKey);
 119            else
 3120                _ = transaction.StringSetAsync(recoveryKey, AsyncResponseJson.Serialize(states), Expiration.KeepTtl, Val
 3121            if (await transaction.ExecuteAsync().ConfigureAwait(false))
 3122                return true;
 123        }
 124
 125        // Leave the registration for expiry rather than risking a lost concurrent registration
 126        // with an unconditional rewrite; the caller treats false as "nothing deleted".
 3127        _logger.LogWarning(
 3128            "Recovery-state delete for correlationId {CorrelationId} registration {RegistrationId} exhausted {Attempts} 
 3129            correlationId, registrationId, MaxCasAttempts);
 2130        return false;
 3131    }
 132
 133    /// <inheritdoc />
 134    public async IAsyncEnumerable<RecoveryState> ScanAsync([EnumeratorCancellation] CancellationToken cancellationToken 
 135    {
 3136        var connectedServers = _multiplexer.GetEndPoints()
 3137            .Select(endPoint => _multiplexer.GetServer(endPoint))
 3138            .Where(server => server.IsConnected)
 3139            .ToList();
 140
 3141        var seenKeys = new HashSet<string>(StringComparer.Ordinal);
 142
 3143        foreach (var server in connectedServers)
 144        {
 3145            foreach (var key in server.Keys(pattern: _keys.RecoveryKeyPattern, pageSize: 250))
 146            {
 3147                cancellationToken.ThrowIfCancellationRequested();
 148
 3149                var recoveryKey = key.ToString();
 3150                if (!seenKeys.Add(recoveryKey))
 151                    continue;
 152
 3153                var value = await _database.StringGetAsync(recoveryKey).ConfigureAwait(false);
 3154                if (value.IsNullOrEmpty)
 155                    continue;
 156
 3157                var correlationId = _keys.CorrelationIdFromRecoveryKey(recoveryKey);
 3158                foreach (var state in DeserializeStates(value, recoveryKey, correlationId, logAsError: false))
 3159                    yield return state;
 3160            }
 161        }
 3162    }
 163
 164    private const int MaxCasAttempts = 4;
 165
 166    private async Task<List<RecoveryState>> LoadStatesAsync(string recoveryKey, string correlationId)
 167    {
 3168        var value = await _database.StringGetAsync(recoveryKey).ConfigureAwait(false);
 3169        return value.IsNullOrEmpty
 3170            ? []
 3171            : DeserializeStates(value, recoveryKey, correlationId, logAsError: true);
 3172    }
 173
 174    private List<RecoveryState> DeserializeStates(RedisValue value, string recoveryKey, string correlationId, bool logAs
 175    {
 176        try
 177        {
 3178            var states = AsyncResponseJson.Deserialize<List<RecoveryState>>(value.ToString()) ?? [];
 179
 3180            for (var i = states.Count - 1; i >= 0; i--)
 181            {
 3182                var state = states[i];
 3183                if (state is null || state.RegistrationId == Guid.Empty)
 184                {
 3185                    _logger.LogWarning(
 3186                        "Recovery state at {RecoveryKey} has no registration id; rejecting it because it cannot be delet
 3187                        recoveryKey);
 2188                    states.RemoveAt(i);
 3189                    continue;
 190                }
 191
 3192                if (!RecoveryStateSchema.IsReadable(state.SchemaVersion))
 193                {
 3194                    _logger.LogWarning(
 3195                        "Recovery state at {RecoveryKey} has unsupported schema version {SchemaVersion} (current: {Curre
 3196                        recoveryKey, state.SchemaVersion, RecoveryStateSchema.Current);
 2197                    states.RemoveAt(i);
 3198                    continue;
 199                }
 200
 3201                if (!string.Equals(state.CorrelationId, correlationId, StringComparison.Ordinal))
 202                {
 3203                    _logger.LogWarning(
 3204                        "Recovery state at {RecoveryKey} has correlationId {StoredCorrelationId}, expected {CorrelationI
 3205                        recoveryKey, state.CorrelationId, correlationId);
 3206                    states.RemoveAt(i);
 207                }
 208            }
 209
 3210            return states;
 211        }
 3212        catch (JsonException ex)
 213        {
 3214            if (logAsError)
 2215                _logger.LogError(ex, "Failed to deserialize recovery state at {RecoveryKey}.", recoveryKey);
 216            else
 2217                _logger.LogWarning(ex, "Unreadable recovery state at {RecoveryKey}; skipping.", recoveryKey);
 3218            return [];
 219        }
 3220    }
 221
 222}
Generated by: ReportGenerator 5.5.10.0
08/03/2026 - 08:59:44
GitHub | reportgenerator.io

Methods/Properties

.ctor(StackExchange.Redis.IConnectionMultiplexer, Microsoft.Extensions.Options.IOptions<AsyncResponse.Channels.Redis.RedisAsyncResponseOptions>, Microsoft.Extensions.Logging.ILogger<AsyncResponse.Channels.Redis.RedisRecoveryStateStore>)
SaveAsync()
GetAllAsync()
TryDeleteAsync()
ScanAsync()
LoadStatesAsync()
DeserializeStates(StackExchange.Redis.RedisValue, string, string, bool)