| | | 1 | | using Microsoft.Extensions.Logging; |
| | | 2 | | using Microsoft.Extensions.Options; |
| | | 3 | | using StackExchange.Redis; |
| | | 4 | | using System.Runtime.CompilerServices; |
| | | 5 | | using System.Text.Json; |
| | | 6 | | using System.Text.Json.Serialization; |
| | | 7 | | using System.Text.Json.Serialization.Metadata; |
| | | 8 | | |
| | | 9 | | namespace AsyncResponse.Channels.Redis; |
| | | 10 | | |
| | | 11 | | /// <summary> |
| | | 12 | | /// Redis-backed implementation of <see cref="IRecoveryStateStore"/>. |
| | | 13 | | /// <para> |
| | | 14 | | /// Every registration for a correlation id shares one key, but Redis applies a single TTL per key, |
| | | 15 | | /// so each entry also carries an absolute <see cref="StoredRegistration.ExpiresAtUtc"/>: reads and |
| | | 16 | | /// scans treat an entry past its expiry as absent, saves prune expired entries, and the key TTL is |
| | | 17 | | /// always the longest remaining entry lifetime — a stream of fresh registrations can therefore |
| | | 18 | | /// never keep a dead sibling registration recoverable (nor truncate a longer-lived one). |
| | | 19 | | /// </para> |
| | | 20 | | /// </summary> |
| | | 21 | | internal sealed class RedisRecoveryStateStore : IRecoveryStateStore, IRecoveryStateScanner |
| | | 22 | | { |
| | | 23 | | private readonly IConnectionMultiplexer _multiplexer; |
| | | 24 | | private readonly IDatabase _database; |
| | | 25 | | private readonly RedisKeySchema _keys; |
| | | 26 | | private readonly ILogger<RedisRecoveryStateStore> _logger; |
| | | 27 | | private readonly TimeProvider _timeProvider; |
| | | 28 | | |
| | | 29 | | /// <summary>Creates a Redis-backed recovery state store.</summary> |
| | 524 | 30 | | public RedisRecoveryStateStore( |
| | 524 | 31 | | IConnectionMultiplexer multiplexer, |
| | 524 | 32 | | IOptions<RedisAsyncResponseOptions> options, |
| | 524 | 33 | | ILogger<RedisRecoveryStateStore> logger, |
| | 524 | 34 | | TimeProvider? timeProvider = null) |
| | | 35 | | { |
| | 524 | 36 | | _multiplexer = multiplexer; |
| | 524 | 37 | | _database = multiplexer.GetDatabase(); |
| | 524 | 38 | | _keys = new RedisKeySchema(options.Value.KeyPrefix); |
| | 524 | 39 | | _logger = logger; |
| | 524 | 40 | | _timeProvider = timeProvider ?? TimeProvider.System; |
| | 524 | 41 | | } |
| | | 42 | | |
| | | 43 | | /// <inheritdoc /> |
| | | 44 | | public async Task SaveAsync( |
| | | 45 | | string correlationId, |
| | | 46 | | RecoveryState state, |
| | | 47 | | TimeSpan ttl, |
| | | 48 | | CancellationToken cancellationToken = default) |
| | | 49 | | { |
| | 710 | 50 | | ArgumentException.ThrowIfNullOrWhiteSpace(correlationId); |
| | 708 | 51 | | ArgumentNullException.ThrowIfNull(state); |
| | 706 | 52 | | if (ttl <= TimeSpan.Zero) |
| | 2 | 53 | | throw new ArgumentOutOfRangeException(nameof(ttl), "TTL must be greater than zero."); |
| | 704 | 54 | | if (!string.Equals(state.CorrelationId, correlationId, StringComparison.Ordinal)) |
| | 2 | 55 | | throw new ArgumentException("The recovery-state correlation id must match the store key.", nameof(state)); |
| | 702 | 56 | | if (state.SchemaVersion != RecoveryStateSchema.Current) |
| | 2 | 57 | | throw new ArgumentException("The recovery state must use the current schema version.", nameof(state)); |
| | | 58 | | |
| | 700 | 59 | | cancellationToken.ThrowIfCancellationRequested(); |
| | 698 | 60 | | if (state.RegistrationId == Guid.Empty) |
| | 2 | 61 | | state.RegistrationId = Guid.NewGuid(); |
| | | 62 | | |
| | 698 | 63 | | var recoveryKey = _keys.RecoveryKey(correlationId); |
| | | 64 | | |
| | | 65 | | // Optimistic read-modify-write: two waiters registering the same correlation id |
| | | 66 | | // concurrently must both survive, so each write commits only while the stored value is |
| | | 67 | | // still the one we read (transaction condition), retrying on a conflict. |
| | 1416 | 68 | | for (var attempt = 0; attempt < MaxCasAttempts; attempt++) |
| | | 69 | | { |
| | 706 | 70 | | cancellationToken.ThrowIfCancellationRequested(); |
| | | 71 | | |
| | 706 | 72 | | var nowUtc = _timeProvider.GetUtcNow(); |
| | 706 | 73 | | var previous = await _database.StringGetAsync(recoveryKey).ConfigureAwait(false); |
| | 706 | 74 | | var (entries, legacy) = previous.IsNullOrEmpty |
| | 706 | 75 | | ? (new List<StoredRegistration>(), false) |
| | 706 | 76 | | : DeserializeEntries(previous, recoveryKey, correlationId, logAsError: true, nowUtc, preserveUnreadable: |
| | 704 | 77 | | if (legacy) |
| | | 78 | | { |
| | | 79 | | // Legacy blobs (a bare state list) carry no per-entry expiry — under that format |
| | | 80 | | // every save re-armed the whole key with a full TTL anyway, so re-stamping each |
| | | 81 | | // entry with this save's full TTL preserves that ceiling exactly once; from here |
| | | 82 | | // on the blob is enveloped and expiry is per entry. |
| | 36 | 83 | | foreach (var entry in entries) |
| | 10 | 84 | | entry.ExpiresAtUtc = nowUtc + ttl; |
| | | 85 | | } |
| | | 86 | | |
| | 725 | 87 | | entries.RemoveAll(existing => existing.State?.RegistrationId == state.RegistrationId); |
| | 704 | 88 | | entries.Add(new StoredRegistration { State = state, ExpiresAtUtc = nowUtc + ttl }); |
| | | 89 | | |
| | 704 | 90 | | var transaction = _database.CreateTransaction(); |
| | 704 | 91 | | transaction.AddCondition(previous.IsNull |
| | 704 | 92 | | ? Condition.KeyNotExists(recoveryKey) |
| | 704 | 93 | | : Condition.StringEqual(recoveryKey, previous)); |
| | | 94 | | // The key must outlive its longest-lived entry and no more: a fresh full TTL here |
| | | 95 | | // would re-extend every co-located registration's physical lifetime on each save. |
| | 704 | 96 | | _ = transaction.StringSetAsync(recoveryKey, SerializeEntries(entries), MaxRemaining(entries, nowUtc)); |
| | 704 | 97 | | if (await transaction.ExecuteAsync().ConfigureAwait(false)) |
| | 694 | 98 | | return; |
| | | 99 | | } |
| | | 100 | | |
| | 2 | 101 | | throw new InvalidOperationException( |
| | 2 | 102 | | $"Recovery-state save for correlationId '{correlationId}' could not commit after {MaxCasAttempts} optimistic |
| | 694 | 103 | | } |
| | | 104 | | |
| | | 105 | | /// <inheritdoc /> |
| | | 106 | | public async Task<IReadOnlyList<RecoveryState>> GetAllAsync(string correlationId, CancellationToken cancellationToke |
| | | 107 | | { |
| | 82 | 108 | | ArgumentException.ThrowIfNullOrWhiteSpace(correlationId); |
| | 80 | 109 | | cancellationToken.ThrowIfCancellationRequested(); |
| | | 110 | | |
| | 78 | 111 | | var recoveryKey = _keys.RecoveryKey(correlationId); |
| | 78 | 112 | | return await LoadStatesAsync(recoveryKey, correlationId).ConfigureAwait(false); |
| | 76 | 113 | | } |
| | | 114 | | |
| | | 115 | | /// <inheritdoc /> |
| | | 116 | | public async Task<bool> TryDeleteAsync(string correlationId, Guid registrationId, CancellationToken cancellationToke |
| | | 117 | | { |
| | 428 | 118 | | ArgumentException.ThrowIfNullOrWhiteSpace(correlationId); |
| | 424 | 119 | | if (registrationId == Guid.Empty) |
| | 2 | 120 | | throw new ArgumentException("Registration id cannot be empty.", nameof(registrationId)); |
| | 422 | 121 | | cancellationToken.ThrowIfCancellationRequested(); |
| | | 122 | | |
| | 418 | 123 | | var recoveryKey = _keys.RecoveryKey(correlationId); |
| | | 124 | | |
| | | 125 | | // Optimistic removal: deleting one registration must not clobber a registration that a |
| | | 126 | | // concurrent writer appended between our read and our write. |
| | 856 | 127 | | for (var attempt = 0; attempt < MaxCasAttempts; attempt++) |
| | | 128 | | { |
| | 426 | 129 | | cancellationToken.ThrowIfCancellationRequested(); |
| | | 130 | | |
| | 426 | 131 | | var nowUtc = _timeProvider.GetUtcNow(); |
| | 426 | 132 | | var previous = await _database.StringGetAsync(recoveryKey).ConfigureAwait(false); |
| | 426 | 133 | | if (previous.IsNullOrEmpty) |
| | 2 | 134 | | return false; |
| | | 135 | | |
| | 424 | 136 | | var (entries, legacy) = DeserializeEntries(previous, recoveryKey, correlationId, logAsError: true, nowUtc, p |
| | 867 | 137 | | var removed = entries.RemoveAll(entry => entry.State?.RegistrationId == registrationId) > 0; |
| | 424 | 138 | | if (!removed) |
| | 4 | 139 | | return false; |
| | | 140 | | |
| | 420 | 141 | | var transaction = _database.CreateTransaction(); |
| | 420 | 142 | | transaction.AddCondition(Condition.StringEqual(recoveryKey, previous)); |
| | 420 | 143 | | if (entries.Count == 0) |
| | | 144 | | { |
| | 399 | 145 | | _ = transaction.KeyDeleteAsync(recoveryKey); |
| | | 146 | | } |
| | 21 | 147 | | else if (legacy) |
| | | 148 | | { |
| | | 149 | | // Legacy blobs keep their shape and key TTL here: only SaveAsync migrates to the |
| | | 150 | | // enveloped shape, because it alone has a TTL to stamp the survivors with. |
| | 12 | 151 | | _ = transaction.StringSetAsync( |
| | 12 | 152 | | recoveryKey, |
| | 24 | 153 | | AsyncResponseJson.Serialize(entries.ConvertAll(entry => entry.State!).FindAll(static state => state |
| | 12 | 154 | | Expiration.KeepTtl, |
| | 12 | 155 | | ValueCondition.Always); |
| | | 156 | | } |
| | | 157 | | else |
| | | 158 | | { |
| | | 159 | | // Shrink the key to its longest surviving entry: keeping the previous TTL would |
| | | 160 | | // hold the key alive long after the removed registration — possibly the only |
| | | 161 | | // long-lived one — is gone. |
| | 9 | 162 | | _ = transaction.StringSetAsync(recoveryKey, SerializeEntries(entries), MaxRemaining(entries, nowUtc)); |
| | | 163 | | } |
| | 420 | 164 | | if (await transaction.ExecuteAsync().ConfigureAwait(false)) |
| | 410 | 165 | | return true; |
| | | 166 | | } |
| | | 167 | | |
| | | 168 | | // Leave the registration for expiry rather than risking a lost concurrent registration |
| | | 169 | | // with an unconditional rewrite; the caller treats false as "nothing deleted". |
| | 2 | 170 | | _logger.LogWarning( |
| | 2 | 171 | | "Recovery-state delete for correlationId {CorrelationId} registration {RegistrationId} exhausted {Attempts} |
| | 2 | 172 | | correlationId, registrationId, MaxCasAttempts); |
| | 2 | 173 | | return false; |
| | 418 | 174 | | } |
| | | 175 | | |
| | | 176 | | /// <inheritdoc /> |
| | | 177 | | public async IAsyncEnumerable<RecoveryState> ScanAsync([EnumeratorCancellation] CancellationToken cancellationToken |
| | | 178 | | { |
| | 105 | 179 | | var seenKeys = new HashSet<string>(StringComparer.Ordinal); |
| | 105 | 180 | | var batch = new List<string>(ScanReadBatchSize); |
| | | 181 | | |
| | 358 | 182 | | foreach (var server in await ResolveScanTargetsAsync().ConfigureAwait(false)) |
| | | 183 | | { |
| | | 184 | | // KeysAsync, not Keys: the synchronous enumerator blocked its thread on every SCAN |
| | | 185 | | // page of what is, by definition, a walk of the whole keyspace. |
| | 2064 | 186 | | await foreach (var key in server.KeysAsync(pattern: _keys.RecoveryKeyPattern, pageSize: ScanPageSize) |
| | 87 | 187 | | .WithCancellation(cancellationToken) |
| | 87 | 188 | | .ConfigureAwait(false)) |
| | | 189 | | { |
| | 946 | 190 | | cancellationToken.ThrowIfCancellationRequested(); |
| | | 191 | | |
| | 944 | 192 | | var recoveryKey = key.ToString(); |
| | 944 | 193 | | if (!seenKeys.Add(recoveryKey)) |
| | | 194 | | continue; |
| | | 195 | | |
| | 942 | 196 | | batch.Add(recoveryKey); |
| | 942 | 197 | | if (batch.Count < ScanReadBatchSize) |
| | | 198 | | continue; |
| | | 199 | | |
| | 1548 | 200 | | foreach (var state in await ReadScanBatchAsync(batch, cancellationToken).ConfigureAwait(false)) |
| | 768 | 201 | | yield return state; |
| | 6 | 202 | | batch.Clear(); |
| | | 203 | | } |
| | | 204 | | |
| | 85 | 205 | | if (batch.Count == 0) |
| | | 206 | | continue; |
| | | 207 | | |
| | 376 | 208 | | foreach (var state in await ReadScanBatchAsync(batch, cancellationToken).ConfigureAwait(false)) |
| | 164 | 209 | | yield return state; |
| | 23 | 210 | | batch.Clear(); |
| | | 211 | | } |
| | 79 | 212 | | } |
| | | 213 | | |
| | | 214 | | private const int ScanPageSize = 250; |
| | | 215 | | |
| | | 216 | | /// <summary> |
| | | 217 | | /// Values read per pipelined batch. Each registration blob used to be awaited before the next |
| | | 218 | | /// GET was even sent — one network round trip per key, so a 100,000-key scan at 2 ms spent |
| | | 219 | | /// over three minutes on latency alone. The reads are independent, so a batch is issued |
| | | 220 | | /// back to back on the multiplexer's pipeline and awaited together; the bound keeps the scan |
| | | 221 | | /// streaming (the watchdog buffers only the fields it classifies on) instead of holding the |
| | | 222 | | /// keyspace's values in memory. Individual GETs rather than one MGET: on a cluster the keys of |
| | | 223 | | /// a batch hash to different slots, and the multiplexer routes each GET to its own shard. |
| | | 224 | | /// </summary> |
| | | 225 | | private const int ScanReadBatchSize = 128; |
| | | 226 | | |
| | | 227 | | private async Task<List<RecoveryState>> ReadScanBatchAsync(List<string> recoveryKeys, CancellationToken cancellation |
| | | 228 | | { |
| | 31 | 229 | | cancellationToken.ThrowIfCancellationRequested(); |
| | | 230 | | |
| | 31 | 231 | | var reads = new Task<RedisValue>[recoveryKeys.Count]; |
| | 1946 | 232 | | for (var i = 0; i < reads.Length; i++) |
| | 942 | 233 | | reads[i] = _database.StringGetAsync(recoveryKeys[i]); |
| | | 234 | | |
| | | 235 | | // A failed read fails the scan (see ResolveScanTargets): skipping the key would report |
| | | 236 | | // the registrations behind it as absent. |
| | 31 | 237 | | var values = await Task.WhenAll(reads).ConfigureAwait(false); |
| | | 238 | | |
| | 29 | 239 | | var states = new List<RecoveryState>(values.Length); |
| | 29 | 240 | | var nowUtc = _timeProvider.GetUtcNow(); |
| | 1934 | 241 | | for (var i = 0; i < values.Length; i++) |
| | | 242 | | { |
| | 938 | 243 | | if (values[i].IsNullOrEmpty) |
| | | 244 | | continue; |
| | | 245 | | |
| | 936 | 246 | | var recoveryKey = recoveryKeys[i]; |
| | 936 | 247 | | var correlationId = _keys.CorrelationIdFromRecoveryKey(recoveryKey); |
| | 936 | 248 | | var (entries, _) = DeserializeEntries(values[i], recoveryKey, correlationId, logAsError: false, nowUtc); |
| | 3736 | 249 | | foreach (var entry in entries) |
| | 932 | 250 | | states.Add(entry.State!); |
| | | 251 | | } |
| | | 252 | | |
| | 29 | 253 | | return states; |
| | 29 | 254 | | } |
| | | 255 | | |
| | | 256 | | /// <summary> |
| | | 257 | | /// The servers whose keyspaces make up a complete scan — or an exception when that set cannot |
| | | 258 | | /// be inspected. Disconnected servers used to be filtered out silently, so with Redis down |
| | | 259 | | /// the scan "succeeded" over zero servers: the watchdog published an empty report with no |
| | | 260 | | /// error and the recovery health check went from Degraded to Healthy BECAUSE of the outage. |
| | | 261 | | /// An empty keyspace and an unreadable one are different answers; only the first is a scan. |
| | | 262 | | /// <para> |
| | | 263 | | /// Primaries only. Every replica holds a copy of the same keys, so scanning them too walked |
| | | 264 | | /// the keyspace once per node and produced nothing the primary had not already yielded — |
| | | 265 | | /// the dedupe hid the duplicate entries but not the round trips. It also aimed a full |
| | | 266 | | /// keyspace scan at nodes that exist to serve reads cheaply. On a single-node deployment |
| | | 267 | | /// this changes nothing: that node is the primary. |
| | | 268 | | /// </para> |
| | | 269 | | /// </summary> |
| | | 270 | | private async Task<List<IServer>> ResolveScanTargetsAsync() |
| | | 271 | | { |
| | 105 | 272 | | var primaries = new List<IServer>(); |
| | 105 | 273 | | List<IServer>? unreachable = null; |
| | 105 | 274 | | List<IServer>? connectedReplicas = null; |
| | 484 | 275 | | foreach (var endPoint in _multiplexer.GetEndPoints()) |
| | | 276 | | { |
| | 137 | 277 | | var server = _multiplexer.GetServer(endPoint); |
| | 137 | 278 | | if (server.IsReplica) |
| | | 279 | | { |
| | 6 | 280 | | if (server.IsConnected) |
| | 2 | 281 | | (connectedReplicas ??= []).Add(server); |
| | 2 | 282 | | continue; |
| | | 283 | | } |
| | | 284 | | |
| | 131 | 285 | | if (server.IsConnected) |
| | 101 | 286 | | primaries.Add(server); |
| | | 287 | | else |
| | 30 | 288 | | (unreachable ??= []).Add(server); |
| | | 289 | | } |
| | | 290 | | |
| | 105 | 291 | | if (primaries.Count == 0) |
| | | 292 | | { |
| | 8 | 293 | | throw ScanUnavailable( |
| | 8 | 294 | | "Recovery-state scan failed: no Redis primary is connected, so the persisted registrations cannot be rea |
| | 8 | 295 | | "This is an unavailable scan, not an empty keyspace."); |
| | | 296 | | } |
| | | 297 | | |
| | | 298 | | // A cluster shards the keyspace: every primary holds registrations no other primary has, |
| | | 299 | | // so one that cannot be inspected makes the scan partial. Outside a cluster every primary |
| | | 300 | | // the multiplexer knows serves the same dataset (a failed-over deployment lists the old |
| | | 301 | | // primary as disconnected until it rejoins as a replica), and one connected primary is |
| | | 302 | | // the whole keyspace. |
| | 194 | 303 | | if (!primaries.Exists(static primary => primary.ServerType == ServerType.Cluster)) |
| | 79 | 304 | | return primaries; |
| | | 305 | | |
| | | 306 | | // The multiplexer's view of who is a primary is only as good as each node's last |
| | | 307 | | // handshake, in BOTH directions, so a cluster scan is checked against the cluster's own |
| | | 308 | | // node table (CLUSTER NODES — one small command next to a walk of the whole keyspace). |
| | | 309 | | // Conservative in every unknown: a table that cannot be read or lists nothing changes |
| | | 310 | | // nothing below, and the multiplexer's view decides as before. |
| | 18 | 311 | | var nodes = await ReadClusterNodeTableAsync(primaries).ConfigureAwait(false); |
| | | 312 | | |
| | 18 | 313 | | if (unreachable is not null) |
| | | 314 | | { |
| | | 315 | | // A node that has NEVER connected since this process started (a replica that was down |
| | | 316 | | // at startup) still reports the default — not a replica — so it landed here as an |
| | | 317 | | // "unreachable primary" and failed every scan of a cluster whose slot owners were all |
| | | 318 | | // reachable. Only a node that owns slots can make the scan partial; a node the table |
| | | 319 | | // does not list stays unreachable. |
| | 16 | 320 | | if (nodes is not null) |
| | 10 | 321 | | unreachable = unreachable.FindAll(server => !RedisClusterNodeTable.OwnsNoSlots(nodes, server.EndPoint)); |
| | | 322 | | |
| | 16 | 323 | | if (unreachable.Count > 0) |
| | | 324 | | { |
| | 14 | 325 | | throw ScanUnavailable( |
| | 14 | 326 | | $"Recovery-state scan failed: Redis cluster primary {string.Join(", ", unreachable.Select(static ser |
| | 14 | 327 | | "so the registrations in its slots cannot be read. A partial scan is reported as failed rather than |
| | | 328 | | } |
| | | 329 | | } |
| | | 330 | | |
| | 4 | 331 | | if (nodes is not null) |
| | 2 | 332 | | RequireSlotCoverage(nodes, primaries, connectedReplicas); |
| | | 333 | | |
| | 4 | 334 | | return primaries; |
| | 83 | 335 | | } |
| | | 336 | | |
| | | 337 | | private async Task<List<RedisClusterNodeTable.Node>?> ReadClusterNodeTableAsync(List<IServer> primaries) |
| | | 338 | | { |
| | 72 | 339 | | foreach (var primary in primaries) |
| | | 340 | | { |
| | 20 | 341 | | if (primary.ServerType == ServerType.Cluster |
| | 20 | 342 | | && await RedisClusterNodeTable.TryReadAsync(primary, _logger).ConfigureAwait(false) is { } nodes) |
| | 4 | 343 | | return nodes; |
| | | 344 | | } |
| | | 345 | | |
| | 14 | 346 | | return null; |
| | 18 | 347 | | } |
| | | 348 | | |
| | | 349 | | /// <summary> |
| | | 350 | | /// Every slot owner the node table lists must be one of the servers about to be scanned. |
| | | 351 | | /// Excusing the unreachable nodes that own nothing is not the same as knowing every shard is |
| | | 352 | | /// covered: after a failover the old primary is listed WITHOUT slots (excused, correctly), |
| | | 353 | | /// while the promoted node can still carry the multiplexer's pre-failover "replica" flag and |
| | | 354 | | /// be skipped as one — so the scan walked the remaining shards, found them healthy, and |
| | | 355 | | /// reported a complete, possibly empty keyspace with a whole shard's registrations unread. A |
| | | 356 | | /// connected node the table names as a slot owner is scanned whatever the stale flag says; a |
| | | 357 | | /// slot owner with no connected server at all fails the scan, like any other partial one. |
| | | 358 | | /// </summary> |
| | | 359 | | private static void RequireSlotCoverage(List<RedisClusterNodeTable.Node> nodes, List<IServer> primaries, List<IServe |
| | | 360 | | { |
| | 2 | 361 | | List<string>? uncovered = null; |
| | 16 | 362 | | foreach (var node in nodes) |
| | | 363 | | { |
| | 12 | 364 | | if (!node.IsSlotOwner || primaries.Exists(primary => RedisClusterNodeTable.IsSameNode(node, primary.EndPoint |
| | | 365 | | continue; |
| | | 366 | | |
| | 0 | 367 | | if (connectedReplicas?.Find(server => RedisClusterNodeTable.IsSameNode(node, server.EndPoint)) is { } promot |
| | 0 | 368 | | primaries.Add(promoted); |
| | | 369 | | else |
| | 0 | 370 | | (uncovered ??= []).Add($"{node.Address}:{node.Port}"); |
| | | 371 | | } |
| | | 372 | | |
| | 2 | 373 | | if (uncovered is not null) |
| | | 374 | | { |
| | 0 | 375 | | throw ScanUnavailable( |
| | 0 | 376 | | $"Recovery-state scan failed: Redis cluster slot owner {string.Join(", ", uncovered)} has no connected s |
| | 0 | 377 | | "so the registrations in its slots cannot be read. A partial scan is reported as failed rather than as a |
| | | 378 | | } |
| | 2 | 379 | | } |
| | | 380 | | |
| | | 381 | | private static RedisConnectionException ScanUnavailable(string message) |
| | 22 | 382 | | => new(ConnectionFailureType.UnableToConnect, CommandFlags.None, message, innerException: null, CommandStatus.Un |
| | | 383 | | |
| | | 384 | | private const int MaxCasAttempts = 4; |
| | | 385 | | |
| | | 386 | | private async Task<List<RecoveryState>> LoadStatesAsync(string recoveryKey, string correlationId) |
| | | 387 | | { |
| | 78 | 388 | | var value = await _database.StringGetAsync(recoveryKey).ConfigureAwait(false); |
| | 78 | 389 | | if (value.IsNullOrEmpty) |
| | 11 | 390 | | return []; |
| | | 391 | | |
| | 67 | 392 | | var now = _timeProvider.GetUtcNow(); |
| | 67 | 393 | | var (entries, _) = DeserializeEntries(value, recoveryKey, correlationId, logAsError: true, now); |
| | 67 | 394 | | if (entries.Count > 0) |
| | 132 | 395 | | return entries.ConvertAll(entry => entry.State!); |
| | | 396 | | |
| | | 397 | | // The key held a blob and nothing readable came out of it. That must not read as "no |
| | | 398 | | // registration was ever armed" — the dispatcher acknowledges the response on that answer, |
| | | 399 | | // consuming a terminal response whose callback never ran. |
| | | 400 | | // |
| | | 401 | | // Expiry is the exception and has to be told apart here, because DeserializeEntries drops |
| | | 402 | | // lapsed entries by the same route it drops unreadable ones: a registration past its expiry |
| | | 403 | | // is legitimately gone, and failing on it would redeliver forever against a record that is |
| | | 404 | | // supposed to disappear. |
| | 4 | 405 | | if (CountStoredRegistrations(value, out var stored) && stored > 0) |
| | 2 | 406 | | throw new RecoveryStateUnreadableException(correlationId, stored); |
| | | 407 | | |
| | 2 | 408 | | return []; |
| | 76 | 409 | | } |
| | | 410 | | |
| | | 411 | | /// <summary> |
| | | 412 | | /// Counts the registrations physically present in the blob that are NOT past their expiry, |
| | | 413 | | /// without applying the readability rules. The gap between this and what |
| | | 414 | | /// <see cref="DeserializeEntries"/> returned is exactly the unreadable set. Returns |
| | | 415 | | /// <c>false</c> when the blob itself will not parse at all — in which case every registration it |
| | | 416 | | /// held is unreadable by definition, and the caller is told so via <paramref name="stored"/>. |
| | | 417 | | /// </summary> |
| | | 418 | | private bool CountStoredRegistrations(RedisValue value, out int stored) |
| | | 419 | | { |
| | 4 | 420 | | var json = value.ToString(); |
| | 4 | 421 | | var now = _timeProvider.GetUtcNow(); |
| | | 422 | | try |
| | | 423 | | { |
| | 4 | 424 | | if (IsLegacyShape(json)) |
| | | 425 | | { |
| | | 426 | | // Legacy blobs carry no per-entry expiry; every element is a live registration. |
| | 0 | 427 | | stored = JsonSafety.SafeDeserialize(json, _legacyTypeInfo)?.Count ?? 0; |
| | 0 | 428 | | return true; |
| | | 429 | | } |
| | | 430 | | |
| | 4 | 431 | | var parsed = JsonSafety.SafeDeserialize(json, _envelopeTypeInfo); |
| | 2 | 432 | | var registrations = parsed?.Registrations; |
| | 2 | 433 | | if (registrations is null) |
| | | 434 | | { |
| | 2 | 435 | | stored = 0; |
| | 2 | 436 | | return true; |
| | | 437 | | } |
| | | 438 | | |
| | 0 | 439 | | stored = 0; |
| | 0 | 440 | | foreach (var entry in registrations) |
| | | 441 | | { |
| | 0 | 442 | | if (entry is not null && entry.ExpiresAtUtc > now) |
| | 0 | 443 | | stored++; |
| | | 444 | | } |
| | | 445 | | |
| | 0 | 446 | | return true; |
| | | 447 | | } |
| | 2 | 448 | | catch (Exception ex) when (ex is JsonException or InvalidDataException) |
| | | 449 | | { |
| | | 450 | | // Unparseable at the top level: the blob exists and holds an unknown number of |
| | | 451 | | // registrations, all of them unreadable. One is enough to fail the delivery. |
| | 2 | 452 | | stored = 1; |
| | 2 | 453 | | return true; |
| | | 454 | | } |
| | 4 | 455 | | } |
| | | 456 | | |
| | | 457 | | /// <summary> |
| | | 458 | | /// The package-local envelope metadata CHAINED behind the library's resolver, not used alone. |
| | | 459 | | /// A callback argument is <see cref="CallbackParam"/>.Value, typed <c>object</c>, so it |
| | | 460 | | /// serializes by runtime type — and the source generator only emitted what this envelope |
| | | 461 | | /// references transitively (string, int, Guid, DateTime). On its own the context therefore |
| | | 462 | | /// threw NotSupportedException at waiter registration for a perfectly ordinary literal |
| | | 463 | | /// (a bool, a long, an enum, a DTO), on these two channels only, and bypassed |
| | | 464 | | /// AsyncResponseJsonSerialization.RegisterResolver — the documented trim/AOT seam — entirely. |
| | | 465 | | /// The wire format is unchanged: the envelope's own metadata still resolves first. |
| | | 466 | | /// </summary> |
| | 12 | 467 | | private static readonly JsonSerializerOptions _envelopeOptions = new() |
| | 12 | 468 | | { |
| | 12 | 469 | | TypeInfoResolver = JsonTypeInfoResolver.Combine(RedisChannelJsonContext.Default, AsyncResponseJson.Resolver) |
| | 12 | 470 | | }; |
| | | 471 | | |
| | | 472 | | /// <summary>The envelope's metadata off the chained options — the JsonTypeInfo overloads keep this trim/AOT-clean.< |
| | | 473 | | /// <summary>Legacy bare-array blobs, read with the same case-sensitive matching as before.</summary> |
| | 12 | 474 | | private static readonly JsonTypeInfo<List<RecoveryState>> _legacyTypeInfo = |
| | 12 | 475 | | AsyncResponseJson.GetTypeInfo<List<RecoveryState>>(AsyncResponseJson.Default); |
| | | 476 | | |
| | 12 | 477 | | private static readonly JsonTypeInfo<StoredRecoveryState> _envelopeTypeInfo = |
| | 12 | 478 | | AsyncResponseJson.GetTypeInfo<StoredRecoveryState>(_envelopeOptions); |
| | | 479 | | |
| | | 480 | | private static RedisValue SerializeEntries(List<StoredRegistration> entries) |
| | 713 | 481 | | => JsonSerializer.Serialize( |
| | 713 | 482 | | new StoredRecoveryState { Registrations = entries }, |
| | 713 | 483 | | _envelopeTypeInfo); |
| | | 484 | | |
| | | 485 | | private static TimeSpan MaxRemaining(List<StoredRegistration> entries, DateTimeOffset nowUtc) |
| | | 486 | | { |
| | 713 | 487 | | var maxExpiresAtUtc = DateTimeOffset.MinValue; |
| | 2890 | 488 | | foreach (var entry in entries) |
| | | 489 | | { |
| | 732 | 490 | | if (entry.ExpiresAtUtc > maxExpiresAtUtc) |
| | 718 | 491 | | maxExpiresAtUtc = entry.ExpiresAtUtc; |
| | | 492 | | } |
| | | 493 | | |
| | 713 | 494 | | return maxExpiresAtUtc - nowUtc; |
| | | 495 | | } |
| | | 496 | | |
| | | 497 | | /// <summary> |
| | | 498 | | /// Deserializes the stored blob. <c>preserveUnreadable</c>: write paths pass true. A read filters out an entry this |
| | | 499 | | /// schema version, a null state, a blank registration id), but a read-modify-write must carry |
| | | 500 | | /// it through untouched: rewriting the shared blob from the readable subset silently deleted a |
| | | 501 | | /// sibling registration written by a newer host mid-rolling-upgrade — the write path treating |
| | | 502 | | /// "unreadable" as "missing", which is exactly what the read path was hardened to refuse. |
| | | 503 | | /// Expired entries are still pruned either way; those are genuinely gone. |
| | | 504 | | /// </summary> |
| | | 505 | | private (List<StoredRegistration> Entries, bool Legacy) DeserializeEntries( |
| | | 506 | | RedisValue value, |
| | | 507 | | string recoveryKey, |
| | | 508 | | string correlationId, |
| | | 509 | | bool logAsError, |
| | | 510 | | DateTimeOffset nowUtc, |
| | | 511 | | bool preserveUnreadable = false, |
| | | 512 | | bool throwOnUnreadableEnvelope = false) |
| | | 513 | | { |
| | 1450 | 514 | | var json = value.ToString(); |
| | | 515 | | try |
| | | 516 | | { |
| | | 517 | | // Legacy blobs are a bare JSON array of states; enveloped blobs are an object. The |
| | | 518 | | // first significant character tells the shapes apart without a speculative parse. |
| | 1450 | 519 | | if (IsLegacyShape(json)) |
| | | 520 | | { |
| | 106 | 521 | | var states = JsonSafety.SafeDeserialize(json, _legacyTypeInfo) ?? []; |
| | 106 | 522 | | var legacyEntries = new List<StoredRegistration>(states.Count); |
| | 480 | 523 | | foreach (var state in states) |
| | | 524 | | { |
| | | 525 | | // A legacy entry has no expiry of its own; it lives until the key's TTL, as it |
| | | 526 | | // always did (SaveAsync stamps one when it rewrites the blob enveloped). |
| | 134 | 527 | | if (preserveUnreadable || IsStateReadable(state, recoveryKey, correlationId)) |
| | 126 | 528 | | legacyEntries.Add(new StoredRegistration { State = state }); |
| | | 529 | | } |
| | | 530 | | |
| | 106 | 531 | | return (legacyEntries, true); |
| | | 532 | | } |
| | | 533 | | |
| | 1344 | 534 | | var stored = JsonSafety.SafeDeserialize(json, _envelopeTypeInfo); |
| | 1336 | 535 | | var entries = stored?.Registrations ?? []; |
| | 2683 | 536 | | entries.RemoveAll(entry => entry is null || (!preserveUnreadable && !IsStateReadable(entry.State, recoveryKe |
| | | 537 | | // An entry past its per-entry expiry is logically gone even while a longer-lived |
| | | 538 | | // sibling keeps the key alive; surfacing it would fire recovery callbacks for a |
| | | 539 | | // registration that lapsed long ago. |
| | 2683 | 540 | | entries.RemoveAll(entry => entry.ExpiresAtUtc <= nowUtc); |
| | 1336 | 541 | | return (entries, false); |
| | | 542 | | } |
| | 8 | 543 | | catch (Exception ex) when (ex is JsonException or InvalidDataException) |
| | | 544 | | { |
| | | 545 | | // Through JsonSafety, so `ex` is the body-free rebuild (size and position), never the |
| | | 546 | | // reader's own message: that one appends `Path: $.States[0].Context['<key>']` built |
| | | 547 | | // from the stored registration's context keys — tenant and auth baggage — and this |
| | | 548 | | // log line is what carried them into the application log. |
| | 8 | 549 | | if (logAsError) |
| | 6 | 550 | | _logger.LogError(ex, "Failed to deserialize recovery state at {RecoveryKey}.", recoveryKey); |
| | | 551 | | else |
| | 2 | 552 | | _logger.LogWarning(ex, "Unreadable recovery state at {RecoveryKey}; skipping.", recoveryKey); |
| | | 553 | | |
| | | 554 | | // The rewrite path must refuse, not overwrite: an empty result here would make |
| | | 555 | | // SaveAsync commit just the new registration over a blob whose registrations it could |
| | | 556 | | // not even ENUMERATE, destroying every armed callback the blob held — "unreadable" |
| | | 557 | | // read as "missing", which is exactly what the read path was hardened to refuse. |
| | 8 | 558 | | if (throwOnUnreadableEnvelope) |
| | 2 | 559 | | throw new RecoveryStateUnreadableException(correlationId, 1); |
| | | 560 | | |
| | 6 | 561 | | return ([], false); |
| | | 562 | | } |
| | 1448 | 563 | | } |
| | | 564 | | |
| | | 565 | | private static bool IsLegacyShape(string json) |
| | | 566 | | { |
| | 4362 | 567 | | foreach (var ch in json) |
| | | 568 | | { |
| | 1454 | 569 | | if (char.IsWhiteSpace(ch)) |
| | | 570 | | continue; |
| | 1454 | 571 | | return ch == '['; |
| | | 572 | | } |
| | | 573 | | |
| | 0 | 574 | | return false; |
| | | 575 | | } |
| | | 576 | | |
| | | 577 | | private bool IsStateReadable(RecoveryState? state, string recoveryKey, string correlationId) |
| | | 578 | | { |
| | 1013 | 579 | | if (state is null || state.RegistrationId == Guid.Empty) |
| | | 580 | | { |
| | 4 | 581 | | _logger.LogWarning( |
| | 4 | 582 | | "Recovery state at {RecoveryKey} has no registration id; rejecting it because it cannot be deleted safel |
| | 4 | 583 | | recoveryKey); |
| | 4 | 584 | | return false; |
| | | 585 | | } |
| | | 586 | | |
| | 1009 | 587 | | if (!RecoveryStateSchema.IsReadable(state.SchemaVersion)) |
| | | 588 | | { |
| | 2 | 589 | | _logger.LogWarning( |
| | 2 | 590 | | "Recovery state at {RecoveryKey} has unsupported schema version {SchemaVersion} (current: {Current}); re |
| | 2 | 591 | | recoveryKey, state.SchemaVersion, RecoveryStateSchema.Current); |
| | 2 | 592 | | return false; |
| | | 593 | | } |
| | | 594 | | |
| | 1007 | 595 | | if (string.Equals(state.CorrelationId, correlationId, StringComparison.Ordinal)) |
| | 1005 | 596 | | return true; |
| | | 597 | | |
| | 2 | 598 | | _logger.LogWarning( |
| | 2 | 599 | | "Recovery state at {RecoveryKey} has correlationId {StoredCorrelationId}, expected {CorrelationId}; rejectin |
| | 2 | 600 | | recoveryKey, state.CorrelationId, correlationId); |
| | 2 | 601 | | return false; |
| | | 602 | | } |
| | | 603 | | |
| | | 604 | | /// <summary> |
| | | 605 | | /// The stored envelope: an object (legacy blobs were a bare array, which is how the two |
| | | 606 | | /// shapes are told apart) holding every registration with its own absolute expiry. |
| | | 607 | | /// </summary> |
| | | 608 | | internal sealed class StoredRecoveryState |
| | | 609 | | { |
| | 4094 | 610 | | public List<StoredRegistration>? Registrations { get; set; } |
| | | 611 | | } |
| | | 612 | | |
| | | 613 | | /// <summary>One registration and the absolute expiry of the save that wrote it.</summary> |
| | | 614 | | internal sealed class StoredRegistration |
| | | 615 | | { |
| | 5333 | 616 | | public RecoveryState? State { get; set; } |
| | 5590 | 617 | | public DateTimeOffset ExpiresAtUtc { get; set; } |
| | | 618 | | } |
| | | 619 | | } |
| | | 620 | | |
| | | 621 | | /// <summary> |
| | | 622 | | /// Source-generated metadata for the package-local recovery envelope (trim/AOT-safe; Metadata-mode |
| | | 623 | | /// generation with default options serializes the nested <see cref="RecoveryState"/> exactly like |
| | | 624 | | /// the reflection-based path did). |
| | | 625 | | /// </summary> |
| | | 626 | | [JsonSourceGenerationOptions(GenerationMode = JsonSourceGenerationMode.Metadata)] |
| | | 627 | | [JsonSerializable(typeof(RedisRecoveryStateStore.StoredRecoveryState))] |
| | | 628 | | internal sealed partial class RedisChannelJsonContext : JsonSerializerContext; |