| | | 1 | | using Microsoft.Extensions.Logging; |
| | | 2 | | using Microsoft.Extensions.Options; |
| | | 3 | | using System.Runtime.CompilerServices; |
| | | 4 | | using System.Text.Json; |
| | | 5 | | using System.Text.Json.Serialization; |
| | | 6 | | using System.Text.Json.Serialization.Metadata; |
| | | 7 | | |
| | | 8 | | namespace AsyncResponse.Channels.NATS; |
| | | 9 | | |
| | | 10 | | /// <summary> |
| | | 11 | | /// NATS JetStream Key-Value implementation of <see cref="IRecoveryStateStore"/> and |
| | | 12 | | /// <see cref="IRecoveryStateScanner"/>. |
| | | 13 | | /// <para> |
| | | 14 | | /// NATS KV applies a single <c>MaxAge</c> per bucket rather than a TTL per key, so each stored |
| | | 15 | | /// registration carries its own absolute expiry (<see cref="StoredRecoveryState.StateExpiries"/>): |
| | | 16 | | /// reads and scans treat a registration past its stamp as absent — a fresh sibling registration |
| | | 17 | | /// under the same correlation id never extends it — and a fully expired key is deleted |
| | | 18 | | /// best-effort, while the bucket's <c>MaxAge</c> acts as a garbage-collection ceiling for orphans. |
| | | 19 | | /// </para> |
| | | 20 | | /// </summary> |
| | | 21 | | internal sealed class NatsRecoveryStateStore : IRecoveryStateStore, IRecoveryStateScanner |
| | | 22 | | { |
| | | 23 | | private readonly INatsKvStore _store; |
| | | 24 | | private readonly ILogger<NatsRecoveryStateStore> _logger; |
| | | 25 | | private readonly TimeProvider _timeProvider; |
| | | 26 | | |
| | | 27 | | /// <summary>Creates a NATS JetStream Key-Value recovery state store.</summary> |
| | 425 | 28 | | public NatsRecoveryStateStore( |
| | 425 | 29 | | INatsKvStore store, |
| | 425 | 30 | | IOptions<NatsAsyncResponseChannelOptions> options, |
| | 425 | 31 | | ILogger<NatsRecoveryStateStore> logger, |
| | 425 | 32 | | TimeProvider? timeProvider = null) |
| | | 33 | | { |
| | 425 | 34 | | options.Value.Validate(); |
| | 425 | 35 | | _store = store; |
| | 425 | 36 | | _logger = logger; |
| | 425 | 37 | | _timeProvider = timeProvider ?? TimeProvider.System; |
| | 425 | 38 | | } |
| | | 39 | | |
| | | 40 | | /// <inheritdoc /> |
| | | 41 | | public async Task SaveAsync( |
| | | 42 | | string correlationId, |
| | | 43 | | RecoveryState state, |
| | | 44 | | TimeSpan ttl, |
| | | 45 | | CancellationToken cancellationToken = default) |
| | | 46 | | { |
| | 446 | 47 | | ArgumentException.ThrowIfNullOrWhiteSpace(correlationId); |
| | 444 | 48 | | ArgumentNullException.ThrowIfNull(state); |
| | 442 | 49 | | if (ttl <= TimeSpan.Zero) |
| | 2 | 50 | | throw new ArgumentOutOfRangeException(nameof(ttl), "TTL must be greater than zero."); |
| | 440 | 51 | | if (!string.Equals(state.CorrelationId, correlationId, StringComparison.Ordinal)) |
| | 2 | 52 | | throw new ArgumentException("The recovery-state correlation id must match the store key.", nameof(state)); |
| | 438 | 53 | | if (state.SchemaVersion != RecoveryStateSchema.Current) |
| | 2 | 54 | | throw new ArgumentException("The recovery state must use the current schema version.", nameof(state)); |
| | | 55 | | |
| | 436 | 56 | | cancellationToken.ThrowIfCancellationRequested(); |
| | 434 | 57 | | if (state.RegistrationId == Guid.Empty) |
| | 16 | 58 | | state.RegistrationId = Guid.NewGuid(); |
| | | 59 | | |
| | 434 | 60 | | var key = NatsSubjectSchema.RecoveryKey(correlationId); |
| | | 61 | | |
| | | 62 | | // Revision-conditioned read-modify-write: two waiters registering the same correlation id |
| | | 63 | | // concurrently must both survive. |
| | 888 | 64 | | for (var attempt = 0; attempt < MaxCasAttempts; attempt++) |
| | | 65 | | { |
| | 442 | 66 | | cancellationToken.ThrowIfCancellationRequested(); |
| | | 67 | | |
| | 442 | 68 | | var entry = await _store.GetAsync(key, cancellationToken).ConfigureAwait(false); |
| | 442 | 69 | | var stored = entry is { } existing ? TryDeserialize(existing.Value, key) : null; |
| | | 70 | | |
| | | 71 | | // The rewrite path must refuse, not overwrite: an unparseable envelope deserializing |
| | | 72 | | // to "no entries" would make this save commit just the new registration over a blob |
| | | 73 | | // whose registrations it could not even ENUMERATE, destroying every armed callback it |
| | | 74 | | // held — "unreadable" read as "missing", which is exactly what GetAllAsync was |
| | | 75 | | // hardened to refuse. |
| | 442 | 76 | | if (entry is not null && stored is null) |
| | 2 | 77 | | throw new RecoveryStateUnreadableException(correlationId, 1); |
| | | 78 | | |
| | 440 | 79 | | var now = _timeProvider.GetUtcNow(); |
| | 440 | 80 | | List<(RecoveryState? State, DateTimeOffset ExpiresAtUtc)> entries = stored is not null && !IsExpired(stored) |
| | 440 | 81 | | ? EntriesFrom(stored) |
| | 440 | 82 | | : []; |
| | | 83 | | // Deliberately NOT pruned by readability: a registration this build cannot INTERPRET |
| | | 84 | | // (a newer schema version written by a host mid-rolling-upgrade) must be carried |
| | | 85 | | // through untouched. Rewriting the shared envelope from the readable subset silently |
| | | 86 | | // deleted that sibling — the write path treating "unreadable" as "missing", which is |
| | | 87 | | // exactly what GetAllAsync was hardened to refuse. |
| | | 88 | | // |
| | | 89 | | // Per-registration expiry, though, IS pruned: a sibling keeps its OWN stamp rather |
| | | 90 | | // than inheriting this save's fresh TTL. Re-stamping the shared envelope kept a dead |
| | | 91 | | // waiter's registration recoverable for as long as anything else registered under the |
| | | 92 | | // correlation id — its stale callback then fired into a flow that lapsed days |
| | | 93 | | // earlier. An entry past its own stamp is dropped exactly like a relational store's |
| | | 94 | | // per-row expires_at drops it. |
| | 465 | 95 | | entries.RemoveAll(existingEntry => existingEntry.ExpiresAtUtc <= now); |
| | 465 | 96 | | entries.RemoveAll(existingEntry => existingEntry.State is { } existingState && existingState.RegistrationId |
| | 440 | 97 | | entries.Add((state, now + ttl)); |
| | 440 | 98 | | var json = SerializeStates(entries); |
| | | 99 | | |
| | 440 | 100 | | var written = entry is { } current |
| | 440 | 101 | | ? await _store.TryUpdateAsync(key, json, current.Revision, cancellationToken).ConfigureAwait(false) |
| | 440 | 102 | | : await _store.TryCreateAsync(key, json, cancellationToken).ConfigureAwait(false); |
| | 440 | 103 | | if (written) |
| | 430 | 104 | | return; |
| | 10 | 105 | | } |
| | | 106 | | |
| | 2 | 107 | | throw new InvalidOperationException( |
| | 2 | 108 | | $"Recovery-state save for correlationId '{correlationId}' could not commit after {MaxCasAttempts} optimistic |
| | 430 | 109 | | } |
| | | 110 | | |
| | | 111 | | /// <inheritdoc /> |
| | | 112 | | public async Task<IReadOnlyList<RecoveryState>> GetAllAsync(string correlationId, CancellationToken cancellationToke |
| | | 113 | | { |
| | 66 | 114 | | ArgumentException.ThrowIfNullOrWhiteSpace(correlationId); |
| | 64 | 115 | | cancellationToken.ThrowIfCancellationRequested(); |
| | | 116 | | |
| | 64 | 117 | | var key = NatsSubjectSchema.RecoveryKey(correlationId); |
| | 64 | 118 | | var loaded = await LoadStoredAsync(key, cancellationToken).ConfigureAwait(false); |
| | | 119 | | |
| | | 120 | | // The ENVELOPE itself is unreadable — a truncated or corrupt value, or a shape a newer |
| | | 121 | | // build wrote whose JSON this one cannot parse at all (so the per-registration schema check |
| | | 122 | | // below never gets to classify it). That is the same "unreadable is not missing" case the |
| | | 123 | | // per-registration branch guards, one level up: returning [] here would read as "no |
| | | 124 | | // recovery callback was ever armed" and the dispatcher would acknowledge the terminal |
| | | 125 | | // response, consuming it for a callback that never ran. Refuse so redelivery can reach a |
| | | 126 | | // build that can read it. |
| | 64 | 127 | | if (loaded.Unreadable) |
| | 2 | 128 | | throw new RecoveryStateUnreadableException(correlationId, 1); |
| | | 129 | | |
| | 62 | 130 | | if (loaded.Stored is not { } stored) |
| | 13 | 131 | | return []; |
| | | 132 | | |
| | 49 | 133 | | var found = (Stored: stored, loaded.Revision); |
| | | 134 | | |
| | 49 | 135 | | if (IsExpired(found.Stored)) |
| | | 136 | | { |
| | | 137 | | // Past its logical expiry but still physically present (bucket MaxAge has not collected it |
| | | 138 | | // yet): treat as gone and remove it best-effort so it never resurfaces. |
| | 8 | 139 | | await TryDeleteSilentlyAsync(key, found.Revision, cancellationToken).ConfigureAwait(false); |
| | 8 | 140 | | return []; |
| | | 141 | | } |
| | | 142 | | |
| | 41 | 143 | | var now = _timeProvider.GetUtcNow(); |
| | 41 | 144 | | var entries = EntriesFrom(found.Stored); |
| | 41 | 145 | | var states = new List<RecoveryState>(entries.Count); |
| | 41 | 146 | | var unreadable = 0; |
| | 196 | 147 | | foreach (var (state, expiresAtUtc) in entries) |
| | | 148 | | { |
| | | 149 | | // A registration past its own stamp is absence (the relational stores' per-row |
| | | 150 | | // expires_at), even while a fresher sibling keeps the shared key alive. |
| | 57 | 151 | | if (expiresAtUtc <= now) |
| | | 152 | | continue; |
| | | 153 | | |
| | 55 | 154 | | if (!IsStateReadable(state, key, correlationId, ref unreadable)) |
| | | 155 | | continue; |
| | | 156 | | |
| | 41 | 157 | | states.Add(state!); |
| | | 158 | | } |
| | | 159 | | |
| | | 160 | | // Registrations existed and none this build could INTERPRET survived. An empty list reads |
| | | 161 | | // as "no recovery callback was ever armed", which the dispatcher answers by acknowledging |
| | | 162 | | // the response — so a corrupt or newer-schema registration would consume a terminal response |
| | | 163 | | // its callback never saw. A partially readable batch deliberately does not throw (see |
| | | 164 | | // RecoveryStateUnreadableException), and neither does a row rejected for carrying ANOTHER |
| | | 165 | | // correlation id: that row is readable and simply belongs elsewhere, so for the id actually |
| | | 166 | | // asked about it is absence, not corruption. |
| | 41 | 167 | | if (unreadable > 0 && states.Count == 0) |
| | 4 | 168 | | throw new RecoveryStateUnreadableException(correlationId, unreadable); |
| | | 169 | | |
| | 37 | 170 | | return states; |
| | 58 | 171 | | } |
| | | 172 | | |
| | | 173 | | /// <inheritdoc /> |
| | | 174 | | public async Task<bool> TryDeleteAsync(string correlationId, Guid registrationId, CancellationToken cancellationToke |
| | | 175 | | { |
| | 398 | 176 | | ArgumentException.ThrowIfNullOrWhiteSpace(correlationId); |
| | 396 | 177 | | if (registrationId == Guid.Empty) |
| | 2 | 178 | | throw new ArgumentException("Registration id cannot be empty.", nameof(registrationId)); |
| | 394 | 179 | | cancellationToken.ThrowIfCancellationRequested(); |
| | | 180 | | |
| | 394 | 181 | | var key = NatsSubjectSchema.RecoveryKey(correlationId); |
| | | 182 | | |
| | | 183 | | // Revision-conditioned removal: deleting one registration must not clobber a registration |
| | | 184 | | // that another writer appended between our read and our write. |
| | 814 | 185 | | for (var attempt = 0; attempt < MaxCasAttempts; attempt++) |
| | | 186 | | { |
| | 405 | 187 | | cancellationToken.ThrowIfCancellationRequested(); |
| | | 188 | | |
| | 405 | 189 | | var entry = await _store.GetAsync(key, cancellationToken).ConfigureAwait(false); |
| | 405 | 190 | | if (entry is not { } existing) |
| | 4 | 191 | | return false; |
| | | 192 | | |
| | 401 | 193 | | var stored = TryDeserialize(existing.Value, key); |
| | 401 | 194 | | if (stored is null) |
| | 2 | 195 | | return false; |
| | | 196 | | |
| | 399 | 197 | | if (IsExpired(stored)) |
| | | 198 | | { |
| | 2 | 199 | | await TryDeleteSilentlyAsync(key, existing.Revision, cancellationToken).ConfigureAwait(false); |
| | 2 | 200 | | return false; |
| | | 201 | | } |
| | | 202 | | |
| | 397 | 203 | | var entries = EntriesFrom(stored); |
| | | 204 | | // Same rule as SaveAsync: remove only the targeted registration, never a sibling this |
| | | 205 | | // build merely cannot read — dropping those here also let the key be deleted outright |
| | | 206 | | // when they were the only survivors. Entries past their own expiry go too (they are |
| | | 207 | | // already invisible to every read), and the survivors keep their own stamps. |
| | 816 | 208 | | var removed = entries.RemoveAll(candidate => candidate.State is { } candidateState && candidateState.Registr |
| | 397 | 209 | | if (!removed) |
| | 2 | 210 | | return false; |
| | | 211 | | |
| | 417 | 212 | | entries.RemoveAll(candidate => candidate.ExpiresAtUtc <= _timeProvider.GetUtcNow()); |
| | | 213 | | |
| | 395 | 214 | | var succeeded = entries.Count == 0 |
| | 395 | 215 | | ? await _store.TryDeleteAsync(key, existing.Revision, cancellationToken).ConfigureAwait(false) |
| | 395 | 216 | | : await _store.TryUpdateAsync(key, SerializeStates(entries), existing.Revision, cancellationToken).Confi |
| | 395 | 217 | | if (succeeded) |
| | 382 | 218 | | return true; |
| | | 219 | | } |
| | | 220 | | |
| | 2 | 221 | | _logger.LogWarning( |
| | 2 | 222 | | "Recovery-state delete for correlationId {CorrelationId} registration {RegistrationId} exhausted {Attempts} |
| | 2 | 223 | | correlationId, registrationId, MaxCasAttempts); |
| | 2 | 224 | | return false; |
| | 394 | 225 | | } |
| | | 226 | | |
| | | 227 | | /// <inheritdoc /> |
| | | 228 | | public async IAsyncEnumerable<RecoveryState> ScanAsync([EnumeratorCancellation] CancellationToken cancellationToken |
| | | 229 | | { |
| | 564 | 230 | | await foreach (var key in _store.GetKeysAsync(cancellationToken).ConfigureAwait(false)) |
| | | 231 | | { |
| | 12 | 232 | | cancellationToken.ThrowIfCancellationRequested(); |
| | | 233 | | |
| | 12 | 234 | | var loaded = await LoadStoredAsync(key, cancellationToken).ConfigureAwait(false); |
| | | 235 | | |
| | | 236 | | // The watchdog scan reports; it does not settle a delivery, so an unreadable envelope |
| | | 237 | | // is skipped here rather than thrown (TryDeserialize already logged it). GetAllAsync — |
| | | 238 | | // the path whose answer decides whether a terminal response is acknowledged — refuses. |
| | 12 | 239 | | if (loaded.Stored is not { } storedState) |
| | | 240 | | continue; |
| | | 241 | | |
| | 10 | 242 | | var found = (Stored: storedState, loaded.Revision); |
| | | 243 | | |
| | 10 | 244 | | if (IsExpired(found.Stored)) |
| | | 245 | | { |
| | 2 | 246 | | await TryDeleteSilentlyAsync(key, found.Revision, cancellationToken).ConfigureAwait(false); |
| | 2 | 247 | | continue; |
| | | 248 | | } |
| | | 249 | | |
| | 8 | 250 | | var now = _timeProvider.GetUtcNow(); |
| | 36 | 251 | | foreach (var (state, expiresAtUtc) in EntriesFrom(found.Stored)) |
| | | 252 | | { |
| | 10 | 253 | | if (expiresAtUtc <= now) |
| | | 254 | | continue; |
| | | 255 | | |
| | 10 | 256 | | var correlationId = NatsSubjectSchema.CorrelationIdFromRecoveryKey(key); |
| | 10 | 257 | | if (!IsStateReadable(state, key, correlationId)) |
| | | 258 | | continue; |
| | | 259 | | |
| | 6 | 260 | | yield return state!; |
| | | 261 | | } |
| | 8 | 262 | | } |
| | 267 | 263 | | } |
| | | 264 | | |
| | | 265 | | private const int MaxCasAttempts = 4; |
| | | 266 | | |
| | | 267 | | /// <summary> |
| | | 268 | | /// Tri-state load: the key is absent, or it holds an envelope this build can read, or it holds |
| | | 269 | | /// an envelope it cannot. The third case is NOT absence — see the note in <see cref="GetAllAsync"/>. |
| | | 270 | | /// </summary> |
| | | 271 | | private async Task<(StoredRecoveryState? Stored, ulong Revision, bool Unreadable)> LoadStoredAsync(string key, Cance |
| | | 272 | | { |
| | 76 | 273 | | var entry = await _store.GetAsync(key, cancellationToken).ConfigureAwait(false); |
| | 76 | 274 | | if (entry is not { } existing) |
| | 13 | 275 | | return (null, 0, false); |
| | | 276 | | |
| | 63 | 277 | | var stored = TryDeserialize(existing.Value, key); |
| | 63 | 278 | | return stored is null ? (null, existing.Revision, true) : (stored, existing.Revision, false); |
| | 76 | 279 | | } |
| | | 280 | | |
| | | 281 | | /// <summary> |
| | | 282 | | /// The package-local envelope metadata CHAINED behind the library's resolver, not used alone. |
| | | 283 | | /// A callback argument is <see cref="CallbackParam"/>.Value, typed <c>object</c>, so it |
| | | 284 | | /// serializes by runtime type — and the source generator only emitted what this envelope |
| | | 285 | | /// references transitively (string, int, Guid, DateTime). On its own the context therefore |
| | | 286 | | /// threw NotSupportedException at waiter registration for a perfectly ordinary literal |
| | | 287 | | /// (a bool, a long, an enum, a DTO), on this channel and Redis only, and bypassed |
| | | 288 | | /// AsyncResponseJsonSerialization.RegisterResolver — the documented trim/AOT seam — entirely. |
| | | 289 | | /// The wire format is unchanged: the envelope's own metadata still resolves first. |
| | | 290 | | /// </summary> |
| | 13 | 291 | | private static readonly JsonSerializerOptions _envelopeOptions = new() |
| | 13 | 292 | | { |
| | 13 | 293 | | TypeInfoResolver = JsonTypeInfoResolver.Combine(NatsChannelJsonContext.Default, AsyncResponseJson.Resolver) |
| | 13 | 294 | | }; |
| | | 295 | | |
| | | 296 | | /// <summary>The envelope's metadata off the chained options — the JsonTypeInfo overloads keep this trim/AOT-clean.< |
| | 13 | 297 | | private static readonly JsonTypeInfo<StoredRecoveryState> _envelopeTypeInfo = |
| | 13 | 298 | | AsyncResponseJson.GetTypeInfo<StoredRecoveryState>(_envelopeOptions); |
| | | 299 | | |
| | | 300 | | private static string SerializeStates(List<(RecoveryState? State, DateTimeOffset ExpiresAtUtc)> entries) |
| | | 301 | | { |
| | 462 | 302 | | var states = new List<RecoveryState>(entries.Count); |
| | 462 | 303 | | var expiries = new List<DateTimeOffset>(entries.Count); |
| | 462 | 304 | | var maxExpiresAtUtc = DateTimeOffset.MinValue; |
| | 1898 | 305 | | foreach (var (state, expiresAtUtc) in entries) |
| | | 306 | | { |
| | 487 | 307 | | states.Add(state!); |
| | 487 | 308 | | expiries.Add(expiresAtUtc); |
| | 487 | 309 | | if (expiresAtUtc > maxExpiresAtUtc) |
| | 469 | 310 | | maxExpiresAtUtc = expiresAtUtc; |
| | | 311 | | } |
| | | 312 | | |
| | 462 | 313 | | return JsonSerializer.Serialize(new StoredRecoveryState |
| | 462 | 314 | | { |
| | 462 | 315 | | States = states, |
| | 462 | 316 | | StateExpiries = expiries, |
| | 462 | 317 | | ExpiresAtUtc = maxExpiresAtUtc |
| | 462 | 318 | | }, _envelopeTypeInfo); |
| | | 319 | | } |
| | | 320 | | |
| | | 321 | | /// <summary> |
| | | 322 | | /// Pairs each stored registration with its own expiry. Envelopes written before |
| | | 323 | | /// <see cref="StoredRecoveryState.StateExpiries"/> existed carry only the shared stamp, which |
| | | 324 | | /// those registrations inherit — the old behavior, applied to old data only. |
| | | 325 | | /// </summary> |
| | | 326 | | private static List<(RecoveryState? State, DateTimeOffset ExpiresAtUtc)> EntriesFrom(StoredRecoveryState stored) |
| | | 327 | | { |
| | 469 | 328 | | var states = stored.States; |
| | 469 | 329 | | if (states is not { Count: > 0 }) |
| | 4 | 330 | | return []; |
| | | 331 | | |
| | 465 | 332 | | var expiries = stored.StateExpiries is { } perState && perState.Count == states.Count ? stored.StateExpiries : n |
| | 465 | 333 | | var entries = new List<(RecoveryState? State, DateTimeOffset ExpiresAtUtc)>(states.Count); |
| | 1952 | 334 | | for (var i = 0; i < states.Count; i++) |
| | 511 | 335 | | entries.Add((states[i], expiries is not null ? expiries[i] : stored.ExpiresAtUtc)); |
| | | 336 | | |
| | 465 | 337 | | return entries; |
| | | 338 | | } |
| | | 339 | | |
| | 483 | 340 | | private bool IsExpired(StoredRecoveryState stored) => stored.ExpiresAtUtc <= _timeProvider.GetUtcNow(); |
| | | 341 | | |
| | | 342 | | /// <summary>Readability check for paths that prune without judging a read.</summary> |
| | | 343 | | private bool IsStateReadable(RecoveryState? state, string key, string correlationId) |
| | | 344 | | { |
| | 10 | 345 | | var ignored = 0; |
| | 10 | 346 | | return IsStateReadable(state, key, correlationId, ref ignored); |
| | | 347 | | } |
| | | 348 | | |
| | | 349 | | /// <summary> |
| | | 350 | | /// Readability check that also counts rows this build could not INTERPRET. A row carrying |
| | | 351 | | /// another correlation id is deliberately NOT counted: it is readable and belongs elsewhere. |
| | | 352 | | /// </summary> |
| | | 353 | | private bool IsStateReadable(RecoveryState? state, string key, string correlationId, ref int unreadable) |
| | | 354 | | { |
| | 65 | 355 | | if (state is null || state.RegistrationId == Guid.Empty) |
| | | 356 | | { |
| | 4 | 357 | | _logger.LogWarning( |
| | 4 | 358 | | "Recovery state at key {RecoveryKey} has no registration id; rejecting it because it cannot be deleted s |
| | 4 | 359 | | key); |
| | 4 | 360 | | unreadable++; |
| | 4 | 361 | | return false; |
| | | 362 | | } |
| | | 363 | | |
| | 61 | 364 | | if (!RecoveryStateSchema.IsReadable(state.SchemaVersion)) |
| | | 365 | | { |
| | 6 | 366 | | _logger.LogWarning( |
| | 6 | 367 | | "Recovery state at key {RecoveryKey} has unsupported schema version {SchemaVersion} (current: {Current}) |
| | 6 | 368 | | key, state.SchemaVersion, RecoveryStateSchema.Current); |
| | 6 | 369 | | unreadable++; |
| | 6 | 370 | | return false; |
| | | 371 | | } |
| | | 372 | | |
| | 55 | 373 | | if (string.Equals(state.CorrelationId, correlationId, StringComparison.Ordinal)) |
| | 47 | 374 | | return true; |
| | | 375 | | |
| | 8 | 376 | | _logger.LogWarning( |
| | 8 | 377 | | "Recovery state at key {RecoveryKey} has correlationId {StoredCorrelationId}, expected {CorrelationId}; reje |
| | 8 | 378 | | key, state.CorrelationId, correlationId); |
| | 8 | 379 | | return false; |
| | | 380 | | } |
| | | 381 | | |
| | | 382 | | private StoredRecoveryState? TryDeserialize(string json, string key) |
| | | 383 | | { |
| | | 384 | | try |
| | | 385 | | { |
| | | 386 | | // Through JsonSafety, not the raw reader: the exception logged below is the body-free |
| | | 387 | | // rebuild (size and position). The reader's own appends `Path: $.States[0].Context['<key>']` |
| | | 388 | | // built from the stored registration's context keys — tenant and auth baggage — which |
| | | 389 | | // this warning then carried into the application log. |
| | 493 | 390 | | return JsonSafety.SafeDeserialize(json, _envelopeTypeInfo); |
| | | 391 | | } |
| | 10 | 392 | | catch (Exception ex) when (ex is JsonException or InvalidDataException) |
| | | 393 | | { |
| | 10 | 394 | | _logger.LogWarning(ex, "Unreadable recovery state at key {RecoveryKey}; skipping.", key); |
| | 10 | 395 | | return null; |
| | | 396 | | } |
| | 493 | 397 | | } |
| | | 398 | | |
| | | 399 | | private async Task TryDeleteSilentlyAsync(string key, ulong revision, CancellationToken cancellationToken) |
| | | 400 | | { |
| | | 401 | | try |
| | | 402 | | { |
| | | 403 | | // Revision-conditioned like every other write in this store: an unconditional delete |
| | | 404 | | // could destroy a FRESH registration a concurrent SaveAsync committed between our |
| | | 405 | | // expired read and this cleanup. On a revision conflict the delete is simply skipped — |
| | | 406 | | // the new writer owns the key now. |
| | 12 | 407 | | await _store.TryDeleteAsync(key, revision, cancellationToken).ConfigureAwait(false); |
| | 12 | 408 | | } |
| | 0 | 409 | | catch (Exception ex) |
| | | 410 | | { |
| | 0 | 411 | | _logger.LogDebug(ex, "Best-effort delete of expired recovery state at key {RecoveryKey} failed.", key); |
| | 0 | 412 | | } |
| | 12 | 413 | | } |
| | | 414 | | |
| | | 415 | | /// <summary>The stored envelope: the recovery states plus their expiries, for per-registration logical TTL.</summar |
| | | 416 | | internal sealed class StoredRecoveryState |
| | | 417 | | { |
| | 1896 | 418 | | public List<RecoveryState>? States { get; set; } |
| | | 419 | | |
| | | 420 | | /// <summary> |
| | | 421 | | /// Per-registration absolute expiries, parallel to <see cref="States"/> by index. |
| | | 422 | | /// Additive wire property: envelopes written before it existed carry only the shared |
| | | 423 | | /// <see cref="ExpiresAtUtc"/>, which those registrations inherit on read. |
| | | 424 | | /// </summary> |
| | 2323 | 425 | | public List<DateTimeOffset>? StateExpiries { get; set; } |
| | | 426 | | |
| | | 427 | | /// <summary> |
| | | 428 | | /// The envelope-level expiry — the maximum of <see cref="StateExpiries"/>. Kept for the |
| | | 429 | | /// whole-key fast path (every registration expired ⇒ the key is deletable) and for |
| | | 430 | | /// downgrade compatibility with builds that read only this stamp. |
| | | 431 | | /// </summary> |
| | 1940 | 432 | | public DateTimeOffset ExpiresAtUtc { get; set; } |
| | | 433 | | } |
| | | 434 | | } |
| | | 435 | | |
| | | 436 | | /// <summary> |
| | | 437 | | /// Source-generated metadata for the package-local KV envelope (trim/AOT-safe; the wire format is |
| | | 438 | | /// unchanged — Metadata-mode generation with default options matches the previous reflection-based |
| | | 439 | | /// serialization exactly). |
| | | 440 | | /// </summary> |
| | | 441 | | [JsonSourceGenerationOptions(GenerationMode = JsonSourceGenerationMode.Metadata)] |
| | | 442 | | [JsonSerializable(typeof(NatsRecoveryStateStore.StoredRecoveryState))] |
| | | 443 | | internal sealed partial class NatsChannelJsonContext : JsonSerializerContext; |