| | | 1 | | using Microsoft.Extensions.Logging; |
| | | 2 | | using Microsoft.Extensions.Options; |
| | | 3 | | using MongoDB.Bson; |
| | | 4 | | using MongoDB.Bson.Serialization; |
| | | 5 | | using MongoDB.Bson.Serialization.Attributes; |
| | | 6 | | using MongoDB.Bson.Serialization.Serializers; |
| | | 7 | | using MongoDB.Driver; |
| | | 8 | | using System.Runtime.CompilerServices; |
| | | 9 | | using System.Security.Cryptography; |
| | | 10 | | using System.Text; |
| | | 11 | | using AsyncResponse.Internal; |
| | | 12 | | |
| | | 13 | | namespace AsyncResponse.Transports.MongoDB; |
| | | 14 | | |
| | | 15 | | internal enum MongoDbSubscriberRole |
| | | 16 | | { |
| | | 17 | | Worker, |
| | | 18 | | ResponseIngress |
| | | 19 | | } |
| | | 20 | | |
| | | 21 | | /// <summary>A claimed MongoDB transport document, decoupled from driver types for dispatch tests.</summary> |
| | | 22 | | /// <remarks> |
| | | 23 | | /// <c>RenewAsync</c> extends the claim's lease (<c>locked_until</c>) by the original lock timeout, |
| | | 24 | | /// fenced on the claim's <c>lock_id</c>; it returns <c>false</c> when the fence no longer matches |
| | | 25 | | /// (the lease lapsed and another subscriber re-claimed the document). |
| | | 26 | | /// </remarks> |
| | | 27 | | internal sealed record MongoDbTransportDelivery( |
| | | 28 | | Guid Id, |
| | | 29 | | string Queue, |
| | | 30 | | string Payload, |
| | | 31 | | IReadOnlyDictionary<string, string> Headers, |
| | | 32 | | int Attempt, |
| | | 33 | | Func<ValueTask> AckAsync, |
| | | 34 | | Func<TimeSpan, ValueTask> NakAsync, |
| | | 35 | | Func<Exception, bool, CancellationToken, ValueTask<bool>> DeadLetterAsync, |
| | | 36 | | Func<ValueTask<bool>> RenewAsync); |
| | | 37 | | |
| | | 38 | | /// <summary>Small document adapter for the MongoDB transport queue collection.</summary> |
| | | 39 | | internal sealed class MongoDbTransportStore : IDisposable |
| | | 40 | | { |
| | | 41 | | private readonly IMongoCollection<MongoTransportMessageDocument> _messages; |
| | | 42 | | private readonly MongoDbAsyncResponseTransportOptions _options; |
| | | 43 | | private readonly ILogger<MongoDbTransportStore>? _logger; |
| | 263 | 44 | | private readonly SemaphoreSlim _ensureGate = new(1, 1); |
| | | 45 | | private readonly IMongoClient? _ownedClient; |
| | | 46 | | private bool _created; |
| | | 47 | | private long _lastDeadLetterPruneTicks; |
| | | 48 | | private readonly IMongoDatabase _database; |
| | | 49 | | |
| | 263 | 50 | | public MongoDbTransportStore( |
| | 263 | 51 | | IMongoDatabase database, |
| | 263 | 52 | | IOptions<MongoDbAsyncResponseTransportOptions> options, |
| | 263 | 53 | | IMongoClient? ownedClient = null, |
| | 263 | 54 | | ILogger<MongoDbTransportStore>? logger = null, |
| | 263 | 55 | | IMongoNamespaceRegistry? namespaceRegistry = null) |
| | | 56 | | { |
| | 263 | 57 | | _options = options.Value; |
| | 263 | 58 | | _logger = logger; |
| | 263 | 59 | | MongoDbTransportOptionsValidator.ValidateCommon(_options); |
| | | 60 | | |
| | | 61 | | // Cross-component collection ownership (DI-hosted stores only): see MongoNamespaceRegistry. |
| | 263 | 62 | | namespaceRegistry?.Claim( |
| | 263 | 63 | | MongoNamespaceRegistry.ClusterKey(database), |
| | 263 | 64 | | database.DatabaseNamespace.DatabaseName, |
| | 263 | 65 | | "MongoDB transport", |
| | 263 | 66 | | [(_options.MessageCollection, nameof(_options.MessageCollection))]); |
| | | 67 | | |
| | | 68 | | // The namespace BYTE limit can only be checked here, where the actual database name is |
| | | 69 | | // first known; a near-limit configuration otherwise passes every static check and fails |
| | | 70 | | // at the first server operation. |
| | 261 | 71 | | MongoNamespaceRegistry.ValidateEffectiveNamespace(database, _options.MessageCollection, nameof(_options.MessageC |
| | | 72 | | |
| | 257 | 73 | | _database = database; |
| | | 74 | | // Pinned to the primary (channel / flow-store parity): a secondaryPreferred client would |
| | | 75 | | // route the change-stream wake to a lagging secondary, so worker jobs woke at replication |
| | | 76 | | // lag and delivery quietly degraded to EmptyPollDelay polling. |
| | 257 | 77 | | _messages = database.GetCollection<MongoTransportMessageDocument>(_options.MessageCollection) |
| | 257 | 78 | | .WithReadPreference(ReadPreference.Primary); |
| | 257 | 79 | | _ownedClient = ownedClient; |
| | 257 | 80 | | } |
| | | 81 | | |
| | | 82 | | public async Task EnsureCreatedAsync(CancellationToken cancellationToken = default) |
| | | 83 | | { |
| | | 84 | | // Not short-circuited on AutoCreateIndexes/UseOwnershipLedger both being off: the |
| | | 85 | | // read-only index check below still runs once, and _created then keeps this cheap. |
| | 3157 | 86 | | if (_created) |
| | 2535 | 87 | | return; |
| | | 88 | | |
| | 622 | 89 | | await _ensureGate.WaitAsync(cancellationToken).ConfigureAwait(false); |
| | | 90 | | try |
| | | 91 | | { |
| | 622 | 92 | | if (_created) |
| | 384 | 93 | | return; |
| | | 94 | | |
| | | 95 | | // Persisted cross-host ownership, independent of AutoCreateIndexes: see |
| | | 96 | | // MongoOwnershipLedger. |
| | 238 | 97 | | if (_options.UseOwnershipLedger) |
| | | 98 | | { |
| | 234 | 99 | | await MongoOwnershipLedger.ClaimAsync( |
| | 234 | 100 | | _database, |
| | 234 | 101 | | "MongoDB transport", |
| | 234 | 102 | | [(_options.MessageCollection, nameof(_options.MessageCollection))], |
| | 234 | 103 | | cancellationToken).ConfigureAwait(false); |
| | | 104 | | } |
| | | 105 | | |
| | 238 | 106 | | if (!_options.AutoCreateIndexes) |
| | | 107 | | { |
| | | 108 | | // Channel / flow-store parity: verify (read-only, warn-only) instead of skipping |
| | | 109 | | // silently. A missing claim index turned every claim into a full collection scan |
| | | 110 | | // per poll tick, per subscriber, with no error and no log line. |
| | 44 | 111 | | await WarnIfClaimIndexMissingAsync(cancellationToken).ConfigureAwait(false); |
| | 44 | 112 | | _created = true; |
| | 44 | 113 | | return; |
| | | 114 | | } |
| | | 115 | | |
| | 194 | 116 | | await _messages.Indexes.CreateOneAsync( |
| | 194 | 117 | | new CreateIndexModel<MongoTransportMessageDocument>( |
| | 194 | 118 | | Builders<MongoTransportMessageDocument>.IndexKeys |
| | 194 | 119 | | .Ascending(item => item.Queue) |
| | 194 | 120 | | .Ascending(item => item.AvailableAtUtc) |
| | 194 | 121 | | .Ascending(item => item.CreatedAtUtc), |
| | 194 | 122 | | new CreateIndexOptions { Name = $"{_options.MessageCollection}_claim_idx" }), |
| | 194 | 123 | | cancellationToken: cancellationToken).ConfigureAwait(false); |
| | 194 | 124 | | await _messages.Indexes.CreateOneAsync( |
| | 194 | 125 | | new CreateIndexModel<MongoTransportMessageDocument>( |
| | 194 | 126 | | Builders<MongoTransportMessageDocument>.IndexKeys.Ascending(item => item.CreatedAtUtc), |
| | 194 | 127 | | new CreateIndexOptions { Name = $"{_options.MessageCollection}_created_idx" }), |
| | 194 | 128 | | cancellationToken: cancellationToken).ConfigureAwait(false); |
| | 194 | 129 | | _created = true; |
| | 194 | 130 | | } |
| | | 131 | | finally |
| | | 132 | | { |
| | 622 | 133 | | _ensureGate.Release(); |
| | | 134 | | } |
| | 3157 | 135 | | } |
| | | 136 | | |
| | | 137 | | private async Task WarnIfClaimIndexMissingAsync(CancellationToken cancellationToken) |
| | | 138 | | { |
| | | 139 | | try |
| | | 140 | | { |
| | | 141 | | List<BsonDocument> indexes; |
| | | 142 | | try |
| | | 143 | | { |
| | 44 | 144 | | using var cursor = await _messages.Indexes.ListAsync(cancellationToken).ConfigureAwait(false); |
| | 2 | 145 | | indexes = await cursor.ToListAsync(cancellationToken).ConfigureAwait(false); |
| | 2 | 146 | | } |
| | 0 | 147 | | catch (MongoCommandException ex) when (ex.Code == 26) |
| | | 148 | | { |
| | | 149 | | // NamespaceNotFound: MongoDB creates the collection bare on the first write — |
| | | 150 | | // which, with index DDL disabled, is exactly a collection with no claim index. |
| | 0 | 151 | | indexes = []; |
| | 0 | 152 | | } |
| | | 153 | | |
| | | 154 | | // Matched by KEY, not by name: operators own the naming of manually provisioned indexes. |
| | 2 | 155 | | var claimIndexed = indexes.Any(index => |
| | 2 | 156 | | index.TryGetValue("key", out var key) |
| | 2 | 157 | | && key is BsonDocument keys |
| | 2 | 158 | | && keys.ElementCount > 0 |
| | 2 | 159 | | && string.Equals(keys.GetElement(0).Name, "queue", StringComparison.Ordinal)); |
| | 2 | 160 | | if (!claimIndexed) |
| | | 161 | | { |
| | 2 | 162 | | _logger?.LogWarning( |
| | 2 | 163 | | "MongoDB collection {Database}.{Collection} has no index leading on 'queue' and AutoCreateIndexes is |
| | 2 | 164 | | "Every claim scans the whole collection on every poll tick — performance only; create the claim inde |
| | 2 | 165 | | "(queue, available_at, created_at) or enable AutoCreateIndexes.", |
| | 2 | 166 | | _database.DatabaseNamespace.DatabaseName, |
| | 2 | 167 | | _options.MessageCollection); |
| | | 168 | | } |
| | 2 | 169 | | } |
| | 42 | 170 | | catch (Exception ex) when (ex is not OperationCanceledException) |
| | | 171 | | { |
| | | 172 | | // A deployment that cannot even list indexes (no listIndexes privilege, server |
| | | 173 | | // unreachable at first use) must not lose the actual operation to the check: the |
| | | 174 | | // caller's own store call surfaces any real connectivity failure. |
| | 42 | 175 | | _logger?.LogDebug(ex, "Skipping index verification for the manually managed MongoDB transport collection; li |
| | 42 | 176 | | } |
| | 44 | 177 | | } |
| | | 178 | | |
| | | 179 | | /// <summary> |
| | | 180 | | /// Publishes a queue document. The caller supplies the id so a retried publish is idempotent — |
| | | 181 | | /// a duplicate-key insert is treated as success rather than enqueuing the same job twice. |
| | | 182 | | /// </summary> |
| | | 183 | | public async Task PublishAsync( |
| | | 184 | | Guid id, |
| | | 185 | | string queue, |
| | | 186 | | string payload, |
| | | 187 | | IReadOnlyDictionary<string, string>? headers, |
| | | 188 | | CancellationToken cancellationToken, |
| | | 189 | | TimeSpan? delay = null) |
| | | 190 | | { |
| | 423 | 191 | | await InsertAsync(id, queue, payload, headers, deadLetterReason: null, cancellationToken, delay).ConfigureAwait( |
| | 421 | 192 | | await PruneDeadLettersIfDueAsync(cancellationToken).ConfigureAwait(false); |
| | 421 | 193 | | } |
| | | 194 | | |
| | | 195 | | public async Task<MongoDbTransportDelivery?> TryClaimAsync(string queue, TimeSpan lockTimeout, CancellationToken can |
| | | 196 | | { |
| | 1911 | 197 | | await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); |
| | 1911 | 198 | | var lockId = Guid.NewGuid(); |
| | | 199 | | |
| | | 200 | | // findOneAndUpdate is atomic per document: of all competing consumers, exactly one observes |
| | | 201 | | // the document unlocked and stamps its lock_id/locked_until in the same server-side step. |
| | | 202 | | // Binary (simple) collation pinned on the claim (SQL Server BIN2 / PostgreSQL |
| | | 203 | | // deterministic-collation parity): the three logical queues share this collection and |
| | | 204 | | // are told apart by nothing but the queue field, and an operator-created collection |
| | | 205 | | // with a case- or accent-folding default collation made the worker subscriber claim |
| | | 206 | | // response documents — which the ingress then dropped and ACKed with no dead-letter |
| | | 207 | | // record. An explicit simple collation cannot use an index built with a folding one, so |
| | | 208 | | // correctness costs a scan there; the default (bare) collection is unaffected. |
| | 1911 | 209 | | var claimed = await _messages.FindOneAndUpdateAsync( |
| | 1911 | 210 | | BuildClaimFilter(queue), |
| | 1911 | 211 | | BuildClaimUpdate(lockId, lockTimeout), |
| | 1911 | 212 | | new FindOneAndUpdateOptions<MongoTransportMessageDocument> |
| | 1911 | 213 | | { |
| | 1911 | 214 | | Sort = Builders<MongoTransportMessageDocument>.Sort.Ascending(item => item.CreatedAtUtc), |
| | 1911 | 215 | | ReturnDocument = ReturnDocument.After, |
| | 1911 | 216 | | Collation = Collation.Simple |
| | 1911 | 217 | | }, |
| | 1911 | 218 | | cancellationToken).ConfigureAwait(false); |
| | 1845 | 219 | | if (claimed is null) |
| | 1407 | 220 | | return null; |
| | | 221 | | |
| | | 222 | | // Indexer, not the copying constructor: documents can be written by foreign producers, and |
| | | 223 | | // BSON legally carries field names differing only in case — the constructor's internal Add |
| | | 224 | | // would throw AFTER the claim already stamped attempts+1/lock_id, before any delivery |
| | | 225 | | // exists, so the document could never reach HandleFailureAsync or dead-letter: an |
| | | 226 | | // unkillable poison document that tears down the subscriber on every re-claim. Last-wins, |
| | | 227 | | // matching the ASB/SQS receive adapters. |
| | | 228 | | IReadOnlyDictionary<string, string> headers; |
| | 438 | 229 | | if (claimed.Headers is null) |
| | | 230 | | { |
| | 10 | 231 | | headers = EmptyHeaders; |
| | | 232 | | } |
| | | 233 | | else |
| | | 234 | | { |
| | 428 | 235 | | var copied = new Dictionary<string, string>(claimed.Headers.Count, StringComparer.OrdinalIgnoreCase); |
| | 1074 | 236 | | foreach (var pair in claimed.Headers) |
| | 109 | 237 | | copied[pair.Key] = pair.Value; |
| | 428 | 238 | | headers = copied; |
| | | 239 | | } |
| | | 240 | | |
| | 438 | 241 | | return new MongoDbTransportDelivery( |
| | 438 | 242 | | claimed.Id, |
| | 438 | 243 | | claimed.Queue, |
| | 438 | 244 | | claimed.Payload, |
| | 438 | 245 | | headers, |
| | 438 | 246 | | claimed.Attempts, |
| | 417 | 247 | | () => AckAsync(claimed.Id, lockId), |
| | 6 | 248 | | delay => NakAsync(claimed.Id, lockId, delay), |
| | 9 | 249 | | (exception, deleteOriginal, token) => DeadLetterAsync(claimed.Id, lockId, queue, claimed.Payload, headers, e |
| | 438 | 250 | | () => RenewLeaseAsync(claimed.Id, lockId, lockTimeout)); |
| | 1845 | 251 | | } |
| | | 252 | | |
| | | 253 | | /// <summary> |
| | | 254 | | /// Claim filter: available and not (still) locked, evaluated against the server clock |
| | | 255 | | /// (<c>$$NOW</c>) so publisher/consumer clock skew never fences messages in or out. |
| | | 256 | | /// A missing <c>locked_until</c> compares as null and therefore as expired. |
| | | 257 | | /// </summary> |
| | | 258 | | internal static FilterDefinition<MongoTransportMessageDocument> BuildClaimFilter(string queue) |
| | 1913 | 259 | | => Builders<MongoTransportMessageDocument>.Filter.Eq(item => item.Queue, queue) |
| | 1913 | 260 | | & new BsonDocumentFilterDefinition<MongoTransportMessageDocument>(new BsonDocument( |
| | 1913 | 261 | | "$expr", |
| | 1913 | 262 | | new BsonDocument("$and", new BsonArray |
| | 1913 | 263 | | { |
| | 1913 | 264 | | new BsonDocument("$lte", new BsonArray { "$available_at", "$$NOW" }), |
| | 1913 | 265 | | new BsonDocument("$or", new BsonArray |
| | 1913 | 266 | | { |
| | 1913 | 267 | | new BsonDocument("$eq", new BsonArray { "$locked_until", BsonNull.Value }), |
| | 1913 | 268 | | new BsonDocument("$lte", new BsonArray { "$locked_until", "$$NOW" }) |
| | 1913 | 269 | | }) |
| | 1913 | 270 | | }))); |
| | | 271 | | |
| | | 272 | | internal static UpdateDefinition<MongoTransportMessageDocument> BuildClaimUpdate(Guid lockId, TimeSpan lockTimeout) |
| | 1913 | 273 | | => Builders<MongoTransportMessageDocument>.Update.Pipeline(new[] |
| | 1913 | 274 | | { |
| | 1913 | 275 | | new BsonDocument("$set", new BsonDocument |
| | 1913 | 276 | | { |
| | 1913 | 277 | | ["attempts"] = new BsonDocument("$add", new BsonArray |
| | 1913 | 278 | | { |
| | 1913 | 279 | | new BsonDocument("$ifNull", new BsonArray { "$attempts", 0 }), |
| | 1913 | 280 | | 1 |
| | 1913 | 281 | | }), |
| | 1913 | 282 | | ["locked_until"] = new BsonDocument("$add", new BsonArray { "$$NOW", lockTimeout.TotalMilliseconds }), |
| | 1913 | 283 | | ["lock_id"] = new BsonBinaryData(lockId, GuidRepresentation.Standard) |
| | 1913 | 284 | | }) |
| | 1913 | 285 | | }); |
| | | 286 | | |
| | | 287 | | public async IAsyncEnumerable<MongoDbTransportDelivery> ClaimBatchAsync( |
| | | 288 | | string queue, |
| | | 289 | | int batchSize, |
| | | 290 | | TimeSpan lockTimeout, |
| | | 291 | | [EnumeratorCancellation] CancellationToken cancellationToken) |
| | | 292 | | { |
| | 3784 | 293 | | for (var i = 0; i < batchSize; i++) |
| | | 294 | | { |
| | 1889 | 295 | | var delivery = await TryClaimAsync(queue, lockTimeout, cancellationToken).ConfigureAwait(false); |
| | 1823 | 296 | | if (delivery is null) |
| | 1401 | 297 | | yield break; |
| | 422 | 298 | | yield return delivery; |
| | | 299 | | } |
| | 1404 | 300 | | } |
| | | 301 | | |
| | | 302 | | private async Task InsertAsync( |
| | | 303 | | Guid id, |
| | | 304 | | string queue, |
| | | 305 | | string payload, |
| | | 306 | | IReadOnlyDictionary<string, string>? headers, |
| | | 307 | | string? deadLetterReason, |
| | | 308 | | CancellationToken cancellationToken, |
| | | 309 | | TimeSpan? delay = null) |
| | | 310 | | { |
| | 430 | 311 | | await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); |
| | | 312 | | try |
| | | 313 | | { |
| | 430 | 314 | | await _messages.UpdateOneAsync( |
| | 430 | 315 | | Builders<MongoTransportMessageDocument>.Filter.Eq(item => item.Id, id), |
| | 430 | 316 | | BuildInsertPipeline(queue, payload, headers, deadLetterReason, delay), |
| | 430 | 317 | | new UpdateOptions { IsUpsert = true }, |
| | 430 | 318 | | cancellationToken).ConfigureAwait(false); |
| | 426 | 319 | | } |
| | 0 | 320 | | catch (MongoWriteException ex) when (ex.WriteError?.Category == ServerErrorCategory.DuplicateKey) |
| | | 321 | | { |
| | | 322 | | // Two concurrent upserts of one id can both attempt the insert; the loser's |
| | | 323 | | // duplicate-key error is the outcome the caller asked for (a retried publish found the |
| | | 324 | | // document already present): idempotent success. |
| | 0 | 325 | | } |
| | 426 | 326 | | } |
| | | 327 | | |
| | | 328 | | /// <summary> |
| | | 329 | | /// Upsert pipeline for a queue document. <c>created_at</c> — the claim <c>Sort</c> key and the |
| | | 330 | | /// dead-letter prune cutoff — is stamped from the SERVER clock (<c>$$NOW</c>), matching the |
| | | 331 | | /// claim/renew/NAK updates: a behind-clock publisher's client stamp would otherwise sort its |
| | | 332 | | /// rows permanently to the queue head and shift them across the prune boundary. <c>$ifNull</c> |
| | | 333 | | /// keeps the first (server-stamped) values when a publish retry finds the document already |
| | | 334 | | /// present, and a retry must not reset a claimed document's attempts or lease either — those |
| | | 335 | | /// fields are left alone entirely. |
| | | 336 | | /// </summary> |
| | | 337 | | /// <remarks> |
| | | 338 | | /// "Available immediately on arrival" still stamps epoch: it expresses what the SQL stores' |
| | | 339 | | /// <c>available_at DEFAULT now()</c> expresses without even a same-millisecond tie against the |
| | | 340 | | /// claim filter's <c>$$NOW</c>. A DELAYED publish computes its due time server-relative |
| | | 341 | | /// (<c>$$NOW + delay</c>), mirroring the NAK update, so client clock skew cannot shift it. |
| | | 342 | | /// User-supplied strings ride inside <c>$literal</c>: in a pipeline expression a plain string |
| | | 343 | | /// beginning with <c>$</c> would otherwise be read as a field path or variable. |
| | | 344 | | /// </remarks> |
| | | 345 | | internal static UpdateDefinition<MongoTransportMessageDocument> BuildInsertPipeline( |
| | | 346 | | string queue, |
| | | 347 | | string payload, |
| | | 348 | | IReadOnlyDictionary<string, string>? headers, |
| | | 349 | | string? deadLetterReason, |
| | | 350 | | TimeSpan? delay) |
| | | 351 | | { |
| | 434 | 352 | | var headerArray = new BsonArray(); |
| | 434 | 353 | | if (headers is not null) |
| | | 354 | | { |
| | 452 | 355 | | foreach (var pair in headers) |
| | 117 | 356 | | headerArray.Add(new BsonDocument { ["k"] = pair.Key, ["v"] = pair.Value }); |
| | | 357 | | } |
| | | 358 | | |
| | 434 | 359 | | return Builders<MongoTransportMessageDocument>.Update.Pipeline(new[] |
| | 434 | 360 | | { |
| | 434 | 361 | | new BsonDocument("$set", new BsonDocument |
| | 434 | 362 | | { |
| | 434 | 363 | | ["queue"] = new BsonDocument("$literal", queue), |
| | 434 | 364 | | ["payload"] = new BsonDocument("$literal", payload), |
| | 434 | 365 | | ["headers"] = new BsonDocument("$ifNull", new BsonArray { "$headers", new BsonDocument("$literal", heade |
| | 434 | 366 | | ["created_at"] = new BsonDocument("$ifNull", new BsonArray { "$created_at", "$$NOW" }), |
| | 434 | 367 | | ["available_at"] = new BsonDocument("$ifNull", new BsonArray |
| | 434 | 368 | | { |
| | 434 | 369 | | "$available_at", |
| | 434 | 370 | | delay is { } pending |
| | 434 | 371 | | ? new BsonDocument("$add", new BsonArray { "$$NOW", pending.TotalMilliseconds }) |
| | 434 | 372 | | : (BsonValue)new BsonDateTime(DateTime.UnixEpoch) |
| | 434 | 373 | | }), |
| | 434 | 374 | | ["attempts"] = new BsonDocument("$ifNull", new BsonArray { "$attempts", 0 }), |
| | 434 | 375 | | ["dead_letter_reason"] = new BsonDocument("$ifNull", new BsonArray |
| | 434 | 376 | | { |
| | 434 | 377 | | "$dead_letter_reason", |
| | 434 | 378 | | deadLetterReason is null ? BsonNull.Value : new BsonDocument("$literal", deadLetterReason) |
| | 434 | 379 | | }) |
| | 434 | 380 | | }) |
| | 434 | 381 | | }); |
| | | 382 | | } |
| | | 383 | | |
| | | 384 | | private async ValueTask AckAsync(Guid id, Guid lockId) |
| | | 385 | | { |
| | 419 | 386 | | await _messages.DeleteOneAsync( |
| | 419 | 387 | | Builders<MongoTransportMessageDocument>.Filter.Eq(item => item.Id, id) |
| | 419 | 388 | | & Builders<MongoTransportMessageDocument>.Filter.Eq(item => item.LockId, lockId), |
| | 419 | 389 | | CancellationToken.None).ConfigureAwait(false); |
| | 419 | 390 | | } |
| | | 391 | | |
| | | 392 | | private async ValueTask NakAsync(Guid id, Guid lockId, TimeSpan delay) |
| | | 393 | | { |
| | 6 | 394 | | await _messages.UpdateOneAsync( |
| | 6 | 395 | | Builders<MongoTransportMessageDocument>.Filter.Eq(item => item.Id, id) |
| | 6 | 396 | | & Builders<MongoTransportMessageDocument>.Filter.Eq(item => item.LockId, lockId), |
| | 6 | 397 | | BuildNakUpdate(delay), |
| | 6 | 398 | | options: null, |
| | 6 | 399 | | CancellationToken.None).ConfigureAwait(false); |
| | 6 | 400 | | } |
| | | 401 | | |
| | | 402 | | private async ValueTask<bool> RenewLeaseAsync(Guid id, Guid lockId, TimeSpan lockTimeout) |
| | | 403 | | { |
| | 0 | 404 | | var result = await _messages.UpdateOneAsync( |
| | 0 | 405 | | Builders<MongoTransportMessageDocument>.Filter.Eq(item => item.Id, id) |
| | 0 | 406 | | & Builders<MongoTransportMessageDocument>.Filter.Eq(item => item.LockId, lockId), |
| | 0 | 407 | | BuildRenewUpdate(lockTimeout), |
| | 0 | 408 | | options: null, |
| | 0 | 409 | | CancellationToken.None).ConfigureAwait(false); |
| | 0 | 410 | | return result.MatchedCount > 0; |
| | 0 | 411 | | } |
| | | 412 | | |
| | | 413 | | /// <summary> |
| | | 414 | | /// Fenced lease renewal: extends <c>locked_until</c> from the server clock (<c>$$NOW</c>) only |
| | | 415 | | /// while the claim's <c>lock_id</c> fence still matches, mirroring the claim update. |
| | | 416 | | /// </summary> |
| | | 417 | | internal static UpdateDefinition<MongoTransportMessageDocument> BuildRenewUpdate(TimeSpan lockTimeout) |
| | 2 | 418 | | => Builders<MongoTransportMessageDocument>.Update.Pipeline(new[] |
| | 2 | 419 | | { |
| | 2 | 420 | | new BsonDocument("$set", new BsonDocument |
| | 2 | 421 | | { |
| | 2 | 422 | | ["locked_until"] = new BsonDocument("$add", new BsonArray { "$$NOW", lockTimeout.TotalMilliseconds }) |
| | 2 | 423 | | }) |
| | 2 | 424 | | }); |
| | | 425 | | |
| | | 426 | | internal static UpdateDefinition<MongoTransportMessageDocument> BuildNakUpdate(TimeSpan delay) |
| | 8 | 427 | | => Builders<MongoTransportMessageDocument>.Update.Pipeline(new[] |
| | 8 | 428 | | { |
| | 8 | 429 | | new BsonDocument("$set", new BsonDocument |
| | 8 | 430 | | { |
| | 8 | 431 | | ["available_at"] = new BsonDocument("$add", new BsonArray { "$$NOW", delay.TotalMilliseconds }), |
| | 8 | 432 | | ["locked_until"] = BsonNull.Value, |
| | 8 | 433 | | ["lock_id"] = BsonNull.Value |
| | 8 | 434 | | }) |
| | 8 | 435 | | }); |
| | | 436 | | |
| | | 437 | | private async ValueTask<bool> DeadLetterAsync( |
| | | 438 | | Guid id, |
| | | 439 | | Guid lockId, |
| | | 440 | | string sourceQueue, |
| | | 441 | | string payload, |
| | | 442 | | IReadOnlyDictionary<string, string> headers, |
| | | 443 | | Exception exception, |
| | | 444 | | bool deleteOriginal, |
| | | 445 | | CancellationToken cancellationToken) |
| | | 446 | | { |
| | 9 | 447 | | if (!_options.DeadLetterEnabled) |
| | | 448 | | { |
| | 2 | 449 | | if (deleteOriginal) |
| | 2 | 450 | | await AckAsync(id, lockId).ConfigureAwait(false); |
| | 2 | 451 | | return true; |
| | | 452 | | } |
| | | 453 | | |
| | 7 | 454 | | var deadHeaders = new Dictionary<string, string>(headers, StringComparer.OrdinalIgnoreCase) |
| | 7 | 455 | | { |
| | 7 | 456 | | ["AR-DeadLetter-Reason"] = Sanitize(exception.Message), |
| | 7 | 457 | | ["AR-DeadLetter-Source-Queue"] = sourceQueue |
| | 7 | 458 | | }; |
| | | 459 | | |
| | | 460 | | try |
| | | 461 | | { |
| | 7 | 462 | | if (!deleteOriginal) |
| | | 463 | | { |
| | 3 | 464 | | await InsertAsync(Guid.NewGuid(), _options.DeadLetterQueue, payload, deadHeaders, exception.Message, can |
| | 1 | 465 | | return true; |
| | | 466 | | } |
| | | 467 | | |
| | | 468 | | // MongoDB has no cross-document transaction we can rely on here (the transport must also |
| | | 469 | | // work on standalone servers), so the DLQ insert uses an id derived deterministically |
| | | 470 | | // from the source document: if a crash lands between the insert and the delete, the |
| | | 471 | | // redelivered message dead-letters onto the same id and the duplicate-key insert is |
| | | 472 | | // swallowed — the DLQ never accumulates copies of one poison message. Insert stays |
| | | 473 | | // FIRST for that reason: delete-first would lose the message outright in the same window. |
| | 4 | 474 | | var deadLetterId = DeadLetterId(id); |
| | 4 | 475 | | await InsertAsync(deadLetterId, _options.DeadLetterQueue, payload, deadHeaders, exception.Message, cancellat |
| | | 476 | | |
| | | 477 | | // ...but the burial only counts if the fenced delete matched. A stale claim (the lease |
| | | 478 | | // lapsed and a peer re-claimed the document) must no-op here exactly as the fenced ack |
| | | 479 | | // and NAK do. The DLQ copy written a moment ago is deliberately NOT compensated away: |
| | | 480 | | // the deterministic id means a peer that also reached the cap buried into the SAME |
| | | 481 | | // document, so deleting it here erased the peer's just-logged burial and the message |
| | | 482 | | // vanished from both the live queue and the DLQ. The worst a kept copy can be is a |
| | | 483 | | // spurious DLQ entry for a message whose new owner later succeeds — visible, prunable |
| | | 484 | | // by dead-letter retention, and strictly better than losing the only record. (The SQL |
| | | 485 | | // siblings avoid the dilemma by making delete+insert one atomic statement; standalone |
| | | 486 | | // MongoDB has no equivalent.) |
| | 4 | 487 | | var removed = await _messages.DeleteOneAsync( |
| | 4 | 488 | | Builders<MongoTransportMessageDocument>.Filter.Eq(item => item.Id, id) |
| | 4 | 489 | | & Builders<MongoTransportMessageDocument>.Filter.Eq(item => item.LockId, lockId), |
| | 4 | 490 | | CancellationToken.None).ConfigureAwait(false); |
| | | 491 | | |
| | 4 | 492 | | if (removed.DeletedCount == 0) |
| | | 493 | | { |
| | 2 | 494 | | _logger?.LogWarning( |
| | 2 | 495 | | "MongoDB dead-letter for message {MessageId} from queue {SourceQueue} no-opped: the claim's lease ha |
| | 2 | 496 | | id, |
| | 2 | 497 | | sourceQueue); |
| | 2 | 498 | | return false; |
| | | 499 | | } |
| | | 500 | | |
| | 2 | 501 | | return true; |
| | | 502 | | } |
| | 2 | 503 | | catch (Exception ex) |
| | | 504 | | { |
| | | 505 | | // Callers decide the redelivery consequence from the false return; log the cause here so |
| | | 506 | | // a failing dead-letter write is never silent. |
| | 2 | 507 | | _logger?.LogError( |
| | 2 | 508 | | ex, |
| | 2 | 509 | | "Failed to write MongoDB dead-letter document for message {MessageId} from queue {SourceQueue}.", |
| | 2 | 510 | | id, |
| | 2 | 511 | | sourceQueue); |
| | 2 | 512 | | return false; |
| | | 513 | | } |
| | 9 | 514 | | } |
| | | 515 | | |
| | | 516 | | /// <summary> |
| | | 517 | | /// Watches the queue collection with a change stream and invokes |
| | | 518 | | /// <paramref name="onNotification"/> whenever a document is inserted into |
| | | 519 | | /// <paramref name="queue"/>. Runs until cancellation or a stream error; callers treat the wake |
| | | 520 | | /// as an optimization over <see cref="MongoDbSubscriberOptions.EmptyPollDelay"/> polling. |
| | | 521 | | /// </summary> |
| | | 522 | | public async Task WatchQueueAsync(string queue, Func<Task> onNotification, CancellationToken cancellationToken) |
| | | 523 | | { |
| | 403 | 524 | | await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); |
| | 403 | 525 | | using var cursor = await _messages.WatchAsync( |
| | 403 | 526 | | BuildQueueWatchPipeline(queue), |
| | 403 | 527 | | cancellationToken: cancellationToken).ConfigureAwait(false); |
| | 1165 | 528 | | while (await cursor.MoveNextAsync(cancellationToken).ConfigureAwait(false)) |
| | | 529 | | { |
| | 2248 | 530 | | foreach (var _ in cursor.Current) |
| | 354 | 531 | | await onNotification().ConfigureAwait(false); |
| | | 532 | | } |
| | 4 | 533 | | } |
| | | 534 | | |
| | | 535 | | /// <summary>Change-stream pipeline for queue wakes: a <c>$match</c> on inserts into one logical queue.</summary> |
| | | 536 | | internal static PipelineDefinition<ChangeStreamDocument<MongoTransportMessageDocument>, ChangeStreamDocument<MongoTr |
| | 405 | 537 | | => new EmptyPipelineDefinition<ChangeStreamDocument<MongoTransportMessageDocument>>() |
| | 405 | 538 | | .Match(new BsonDocument("$and", new BsonArray |
| | 405 | 539 | | { |
| | 405 | 540 | | new BsonDocument("operationType", "insert"), |
| | 405 | 541 | | new BsonDocument("fullDocument.queue", queue) |
| | 405 | 542 | | })); |
| | | 543 | | |
| | | 544 | | /// <summary>Returns <c>true</c> when the server rejected the change stream itself (not a transient cursor error).</ |
| | | 545 | | internal static bool IsChangeStreamUnsupported(Exception exception) |
| | 14 | 546 | | => exception is MongoCommandException commandException |
| | 14 | 547 | | && (commandException.Code == 40573 |
| | 14 | 548 | | || commandException.Message.Contains("only supported on replica sets", StringComparison.OrdinalIgnoreCase |
| | | 549 | | |
| | | 550 | | /// <summary> |
| | | 551 | | /// Opportunistically deletes dead-letter documents older than the configured retention. No-op |
| | | 552 | | /// unless <see cref="MongoDbAsyncResponseTransportOptions.DeadLetterRetention"/> is set, and |
| | | 553 | | /// throttled so the delete runs at most once per minute regardless of publish rate. |
| | | 554 | | /// </summary> |
| | | 555 | | private async Task PruneDeadLettersIfDueAsync(CancellationToken cancellationToken) |
| | | 556 | | { |
| | 421 | 557 | | if (_options.DeadLetterRetention is not { } retention || !ShouldPruneDeadLetters()) |
| | 418 | 558 | | return; |
| | | 559 | | |
| | | 560 | | // Same binary collation as the claim: under a folding collection collation this prune |
| | | 561 | | // matched live-queue documents whose name differed only by case. |
| | 3 | 562 | | await _messages.DeleteManyAsync( |
| | 3 | 563 | | BuildDeadLetterPruneFilter(_options.DeadLetterQueue, retention), |
| | 3 | 564 | | new DeleteOptions { Collation = Collation.Simple }, |
| | 3 | 565 | | cancellationToken).ConfigureAwait(false); |
| | 421 | 566 | | } |
| | | 567 | | |
| | | 568 | | /// <summary> |
| | | 569 | | /// Prune filter: age is evaluated ENTIRELY on the server clock — <c>$$NOW</c> against the |
| | | 570 | | /// server-stamped <c>created_at</c> — mirroring the claim filter. Comparing an app-clock |
| | | 571 | | /// cutoff against another instance's stamp mixes two clocks: a pruner running behind the |
| | | 572 | | /// publisher deletes fresh dead letters on arrival, destroying the forensic record of a |
| | | 573 | | /// poison message, and one running ahead keeps them past retention. |
| | | 574 | | /// </summary> |
| | | 575 | | internal static FilterDefinition<MongoTransportMessageDocument> BuildDeadLetterPruneFilter(string deadLetterQueue, T |
| | 5 | 576 | | => Builders<MongoTransportMessageDocument>.Filter.Eq(item => item.Queue, deadLetterQueue) |
| | 5 | 577 | | & new BsonDocumentFilterDefinition<MongoTransportMessageDocument>(new BsonDocument( |
| | 5 | 578 | | "$expr", |
| | 5 | 579 | | new BsonDocument("$lt", new BsonArray |
| | 5 | 580 | | { |
| | 5 | 581 | | "$created_at", |
| | 5 | 582 | | new BsonDocument("$subtract", new BsonArray { "$$NOW", retention.TotalMilliseconds }) |
| | 5 | 583 | | }))); |
| | | 584 | | |
| | | 585 | | private bool ShouldPruneDeadLetters() |
| | | 586 | | { |
| | 10 | 587 | | var now = DateTime.UtcNow.Ticks; |
| | 10 | 588 | | var last = Interlocked.Read(ref _lastDeadLetterPruneTicks); |
| | 10 | 589 | | return now - last >= DeadLetterPruneThrottle.Ticks |
| | 10 | 590 | | && Interlocked.CompareExchange(ref _lastDeadLetterPruneTicks, now, last) == last; |
| | | 591 | | } |
| | | 592 | | |
| | | 593 | | /// <summary> |
| | | 594 | | /// Deterministic dead-letter document id for a source message: the same poison message always |
| | | 595 | | /// maps to the same DLQ id, making the insert-then-delete pair idempotent under crash-redelivery. |
| | | 596 | | /// </summary> |
| | | 597 | | internal static Guid DeadLetterId(Guid sourceId) |
| | | 598 | | { |
| | 15 | 599 | | var hash = SHA256.HashData(Encoding.UTF8.GetBytes($"asyncresponse:deadletter:{sourceId:N}")); |
| | 15 | 600 | | return new Guid(hash.AsSpan(0, 16)); |
| | | 601 | | } |
| | | 602 | | |
| | 7 | 603 | | private static string Sanitize(string value) => value.Replace('\r', ' ').Replace('\n', ' '); |
| | | 604 | | |
| | 3 | 605 | | private static readonly TimeSpan DeadLetterPruneThrottle = TimeSpan.FromMinutes(1); |
| | | 606 | | |
| | 3 | 607 | | private static readonly IReadOnlyDictionary<string, string> EmptyHeaders = |
| | 3 | 608 | | new Dictionary<string, string>(0, StringComparer.OrdinalIgnoreCase); |
| | | 609 | | |
| | | 610 | | /// <summary>Disposes the Mongo client when the store created (and therefore owns) it.</summary> |
| | | 611 | | public void Dispose() |
| | | 612 | | { |
| | 221 | 613 | | _ensureGate.Dispose(); |
| | 221 | 614 | | (_ownedClient as IDisposable)?.Dispose(); |
| | 2 | 615 | | } |
| | | 616 | | } |
| | | 617 | | |
| | | 618 | | /// <remarks> |
| | | 619 | | /// [BsonIgnoreExtraElements] is load-bearing, not tidiness. Queue documents can be written by |
| | | 620 | | /// foreign producers (the documented reply-target consumer) and by newer builds mid-rolling-deploy, |
| | | 621 | | /// and the driver's default is to THROW FormatException for any element outside this class map. |
| | | 622 | | /// That throw lands after findOneAndUpdate has already stamped attempts+1/lock_id and before any |
| | | 623 | | /// delivery object exists, so the document could never reach HandleFailureAsync or the dead-letter |
| | | 624 | | /// queue: it tore the subscriber down on every re-claim, forever, with attempts climbing unbounded. |
| | | 625 | | /// </remarks> |
| | | 626 | | [BsonIgnoreExtraElements] |
| | | 627 | | internal sealed class MongoTransportMessageDocument |
| | | 628 | | { |
| | | 629 | | [BsonId] |
| | | 630 | | [BsonElement("_id")] |
| | | 631 | | [BsonGuidRepresentation(GuidRepresentation.Standard)] |
| | | 632 | | public Guid Id { get; set; } |
| | | 633 | | |
| | | 634 | | [BsonElement("queue")] |
| | | 635 | | public string Queue { get; set; } = ""; |
| | | 636 | | |
| | | 637 | | [BsonElement("payload")] |
| | | 638 | | public string Payload { get; set; } = ""; |
| | | 639 | | |
| | | 640 | | // Array-of-documents representation keeps arbitrary header names (dots, dollars) legal as |
| | | 641 | | // values rather than as BSON field names; the lenient serializer keeps that wire shape while |
| | | 642 | | // never rejecting a foreign producer's value types. |
| | | 643 | | [BsonElement("headers")] |
| | | 644 | | [BsonSerializer(typeof(LenientTransportHeaderSerializer))] |
| | | 645 | | public Dictionary<string, string>? Headers { get; set; } |
| | | 646 | | |
| | | 647 | | [BsonElement("created_at")] |
| | | 648 | | public DateTime CreatedAtUtc { get; set; } |
| | | 649 | | |
| | | 650 | | [BsonElement("available_at")] |
| | | 651 | | public DateTime AvailableAtUtc { get; set; } |
| | | 652 | | |
| | | 653 | | [BsonElement("locked_until")] |
| | | 654 | | public DateTime? LockedUntilUtc { get; set; } |
| | | 655 | | |
| | | 656 | | [BsonElement("lock_id")] |
| | | 657 | | [BsonGuidRepresentation(GuidRepresentation.Standard)] |
| | | 658 | | public Guid? LockId { get; set; } |
| | | 659 | | |
| | | 660 | | [BsonElement("attempts")] |
| | | 661 | | public int Attempts { get; set; } |
| | | 662 | | |
| | | 663 | | [BsonElement("dead_letter_reason")] |
| | | 664 | | public string? DeadLetterReason { get; set; } |
| | | 665 | | } |
| | | 666 | | |
| | | 667 | | /// <summary> |
| | | 668 | | /// Serializes headers in the driver's array-of-documents shape (<c>[{ "k": …, "v": … }, …]</c>) |
| | | 669 | | /// and deserializes them without rejecting ANY BSON a foreign producer can legally store there. |
| | | 670 | | /// The default dictionary serializer throws on a wrong-typed value ("Cannot deserialize a |
| | | 671 | | /// 'String' from BsonType 'Int32'") — and header materialization runs inside the claim's |
| | | 672 | | /// <c>findOneAndUpdate</c>, AFTER the server already stamped <c>attempts+1</c>/<c>lock_id</c> and |
| | | 673 | | /// before any delivery object exists, so that throw could never reach the failure handler or |
| | | 674 | | /// dead-letter: an unkillable poison document that tears down the subscriber on every re-claim. |
| | | 675 | | /// Instead, string values are taken as-is, other scalars keep their canonical (culture-free) |
| | | 676 | | /// string form, document/array values keep their JSON text so correlation extraction still sees a |
| | | 677 | | /// usable string, nulls are skipped, and unusable shapes degrade to no headers — a genuinely |
| | | 678 | | /// poison message then fails in the handler and flows through the NORMAL dead-letter path. |
| | | 679 | | /// </summary> |
| | | 680 | | internal sealed class LenientTransportHeaderSerializer : SerializerBase<Dictionary<string, string>?> |
| | | 681 | | { |
| | | 682 | | public override Dictionary<string, string>? Deserialize(BsonDeserializationContext context, BsonDeserializationArgs |
| | | 683 | | => Materialize(BsonValueSerializer.Instance.Deserialize(context)); |
| | | 684 | | |
| | | 685 | | internal static Dictionary<string, string>? Materialize(BsonValue value) |
| | | 686 | | { |
| | | 687 | | if (value is not BsonArray entries) |
| | | 688 | | return null; |
| | | 689 | | |
| | | 690 | | // Default (ordinal) comparer, matching the driver's own dictionary: the claim-side copy is |
| | | 691 | | // the case-folding point. Indexer writes, so duplicate keys — case-variant or exact, both |
| | | 692 | | // legal BSON — are last-wins instead of a throw. |
| | | 693 | | var headers = new Dictionary<string, string>(entries.Count); |
| | | 694 | | foreach (var entry in entries) |
| | | 695 | | { |
| | | 696 | | if (entry is not BsonDocument pair |
| | | 697 | | || !pair.TryGetValue("k", out var key) |
| | | 698 | | || !pair.TryGetValue("v", out var rawValue)) |
| | | 699 | | { |
| | | 700 | | continue; |
| | | 701 | | } |
| | | 702 | | |
| | | 703 | | var name = Coerce(key); |
| | | 704 | | var text = Coerce(rawValue); |
| | | 705 | | if (name is not null && text is not null) |
| | | 706 | | headers[name] = text; |
| | | 707 | | } |
| | | 708 | | |
| | | 709 | | return headers; |
| | | 710 | | } |
| | | 711 | | |
| | | 712 | | private static string? Coerce(BsonValue value) => value.BsonType switch |
| | | 713 | | { |
| | | 714 | | BsonType.String => value.AsString, |
| | | 715 | | BsonType.Null or BsonType.Undefined => null, |
| | | 716 | | BsonType.Document or BsonType.Array => value.ToJson(), |
| | | 717 | | // The scalar ToString overrides are JSON-flavored and culture-free ("1.5", "123", "true", |
| | | 718 | | // ISO-8601 dates — raw epoch millis when out of DateTime range), and none of them throws. |
| | | 719 | | _ => value.ToString() ?? "" |
| | | 720 | | }; |
| | | 721 | | |
| | | 722 | | public override void Serialize(BsonSerializationContext context, BsonSerializationArgs args, Dictionary<string, stri |
| | | 723 | | { |
| | | 724 | | var writer = context.Writer; |
| | | 725 | | if (value is null) |
| | | 726 | | { |
| | | 727 | | writer.WriteNull(); |
| | | 728 | | return; |
| | | 729 | | } |
| | | 730 | | |
| | | 731 | | writer.WriteStartArray(); |
| | | 732 | | foreach (var pair in value) |
| | | 733 | | { |
| | | 734 | | writer.WriteStartDocument(); |
| | | 735 | | writer.WriteName("k"); |
| | | 736 | | writer.WriteString(pair.Key); |
| | | 737 | | writer.WriteName("v"); |
| | | 738 | | writer.WriteString(pair.Value); |
| | | 739 | | writer.WriteEndDocument(); |
| | | 740 | | } |
| | | 741 | | |
| | | 742 | | writer.WriteEndArray(); |
| | | 743 | | } |
| | | 744 | | } |