| | | 1 | | using Microsoft.Extensions.Options; |
| | | 2 | | using MongoDB.Bson; |
| | | 3 | | using MongoDB.Bson.Serialization.Attributes; |
| | | 4 | | using MongoDB.Driver; |
| | | 5 | | using System.Runtime.CompilerServices; |
| | | 6 | | |
| | | 7 | | namespace AsyncResponse.Channels.MongoDB; |
| | | 8 | | |
| | | 9 | | internal readonly record struct MongoDbChannelMessage( |
| | | 10 | | Guid Id, |
| | | 11 | | string CorrelationId, |
| | | 12 | | string EnvelopeJson, |
| | | 13 | | DateTimeOffset CreatedAtUtc, |
| | | 14 | | DateTimeOffset? AckedAtUtc = null); |
| | | 15 | | |
| | | 16 | | /// <summary>Document adapter for the MongoDB channel collections and change-stream wake.</summary> |
| | | 17 | | internal sealed class MongoDbChannelStore : IDisposable |
| | | 18 | | { |
| | | 19 | | private readonly IMongoCollection<MongoRecoveryStateDocument> _recovery; |
| | | 20 | | private readonly IMongoCollection<MongoChannelMessageDocument> _messages; |
| | | 21 | | private readonly IMongoCollection<MongoChannelSubscriberDocument> _subscribers; |
| | | 22 | | private readonly IMongoDatabase _database; |
| | | 23 | | private readonly MongoDbAsyncResponseChannelOptions _options; |
| | 3 | 24 | | private readonly SemaphoreSlim _ensureGate = new(1, 1); |
| | | 25 | | private readonly IMongoClient? _ownedClient; |
| | | 26 | | private bool _created; |
| | | 27 | | |
| | 3 | 28 | | public MongoDbChannelStore( |
| | 3 | 29 | | IMongoDatabase database, |
| | 3 | 30 | | IOptions<MongoDbAsyncResponseChannelOptions> options, |
| | 3 | 31 | | IMongoClient? ownedClient = null) |
| | | 32 | | { |
| | 3 | 33 | | _options = options.Value; |
| | 3 | 34 | | _options.Validate(); |
| | 3 | 35 | | _database = database; |
| | 3 | 36 | | _recovery = database.GetCollection<MongoRecoveryStateDocument>(_options.RecoveryStateCollection); |
| | 3 | 37 | | _messages = database.GetCollection<MongoChannelMessageDocument>(_options.MessageCollection); |
| | 3 | 38 | | _subscribers = database.GetCollection<MongoChannelSubscriberDocument>(_options.SubscriberCollection); |
| | 3 | 39 | | _ownedClient = ownedClient; |
| | 3 | 40 | | } |
| | | 41 | | |
| | | 42 | | public async Task EnsureCreatedAsync(CancellationToken cancellationToken = default) |
| | | 43 | | { |
| | 3 | 44 | | if (_created || !_options.AutoCreateIndexes) |
| | 3 | 45 | | return; |
| | | 46 | | |
| | 3 | 47 | | await _ensureGate.WaitAsync(cancellationToken).ConfigureAwait(false); |
| | | 48 | | try |
| | | 49 | | { |
| | 3 | 50 | | if (_created) |
| | 3 | 51 | | return; |
| | | 52 | | |
| | | 53 | | // TTL indexes (expireAfterSeconds = 0 on the expiry timestamp) make MongoDB itself reap |
| | | 54 | | // expired documents — no application-side pruning needed. Reads still filter on the |
| | | 55 | | // expiry because the TTL monitor only runs periodically (~60s). |
| | 3 | 56 | | await CreateTtlIndexAsync( |
| | 3 | 57 | | _recovery, |
| | 3 | 58 | | Builders<MongoRecoveryStateDocument>.IndexKeys.Ascending(item => item.ExpiresAtUtc), |
| | 3 | 59 | | $"{_options.RecoveryStateCollection}_expires_idx", |
| | 3 | 60 | | cancellationToken).ConfigureAwait(false); |
| | 3 | 61 | | await _recovery.Indexes.CreateOneAsync( |
| | 3 | 62 | | new CreateIndexModel<MongoRecoveryStateDocument>( |
| | 3 | 63 | | Builders<MongoRecoveryStateDocument>.IndexKeys |
| | 3 | 64 | | .Ascending(item => item.CorrelationId) |
| | 3 | 65 | | .Ascending(item => item.RegisteredAtUtc), |
| | 3 | 66 | | new CreateIndexOptions { Name = $"{_options.RecoveryStateCollection}_correlation_idx" }), |
| | 3 | 67 | | cancellationToken: cancellationToken).ConfigureAwait(false); |
| | | 68 | | |
| | 3 | 69 | | await CreateTtlIndexAsync( |
| | 3 | 70 | | _messages, |
| | 3 | 71 | | Builders<MongoChannelMessageDocument>.IndexKeys.Ascending(item => item.ExpiresAtUtc), |
| | 3 | 72 | | $"{_options.MessageCollection}_expires_idx", |
| | 3 | 73 | | cancellationToken).ConfigureAwait(false); |
| | 3 | 74 | | await _messages.Indexes.CreateOneAsync( |
| | 3 | 75 | | new CreateIndexModel<MongoChannelMessageDocument>( |
| | 3 | 76 | | Builders<MongoChannelMessageDocument>.IndexKeys |
| | 3 | 77 | | .Ascending(item => item.CorrelationId) |
| | 3 | 78 | | .Ascending(item => item.CreatedAtUtc), |
| | 3 | 79 | | new CreateIndexOptions { Name = $"{_options.MessageCollection}_correlation_created_idx" }), |
| | 3 | 80 | | cancellationToken: cancellationToken).ConfigureAwait(false); |
| | | 81 | | |
| | 3 | 82 | | await CreateTtlIndexAsync( |
| | 3 | 83 | | _subscribers, |
| | 3 | 84 | | Builders<MongoChannelSubscriberDocument>.IndexKeys.Ascending(item => item.ExpiresAtUtc), |
| | 3 | 85 | | $"{_options.SubscriberCollection}_expires_idx", |
| | 3 | 86 | | cancellationToken).ConfigureAwait(false); |
| | 3 | 87 | | await _subscribers.Indexes.CreateOneAsync( |
| | 3 | 88 | | new CreateIndexModel<MongoChannelSubscriberDocument>( |
| | 3 | 89 | | Builders<MongoChannelSubscriberDocument>.IndexKeys.Ascending(item => item.CorrelationId), |
| | 3 | 90 | | new CreateIndexOptions { Name = $"{_options.SubscriberCollection}_correlation_idx" }), |
| | 3 | 91 | | cancellationToken: cancellationToken).ConfigureAwait(false); |
| | | 92 | | |
| | 3 | 93 | | _created = true; |
| | 3 | 94 | | } |
| | | 95 | | finally |
| | | 96 | | { |
| | 3 | 97 | | _ensureGate.Release(); |
| | | 98 | | } |
| | 3 | 99 | | } |
| | | 100 | | |
| | | 101 | | private static async Task CreateTtlIndexAsync<TDocument>( |
| | | 102 | | IMongoCollection<TDocument> collection, |
| | | 103 | | IndexKeysDefinition<TDocument> keys, |
| | | 104 | | string indexName, |
| | | 105 | | CancellationToken cancellationToken) |
| | | 106 | | { |
| | 3 | 107 | | var model = new CreateIndexModel<TDocument>( |
| | 3 | 108 | | keys, |
| | 3 | 109 | | new CreateIndexOptions { Name = indexName, ExpireAfter = TimeSpan.Zero }); |
| | | 110 | | try |
| | | 111 | | { |
| | 3 | 112 | | await collection.Indexes.CreateOneAsync(model, cancellationToken: cancellationToken).ConfigureAwait(false); |
| | 3 | 113 | | } |
| | 2 | 114 | | catch (MongoCommandException ex) when (ex.Code is 85 or 86) |
| | | 115 | | { |
| | | 116 | | // IndexOptionsConflict/IndexKeySpecsConflict: an earlier deployment created the |
| | | 117 | | // same-named index with different options. Replace it in place. |
| | 3 | 118 | | await collection.Indexes.DropOneAsync(indexName, cancellationToken).ConfigureAwait(false); |
| | 2 | 119 | | await collection.Indexes.CreateOneAsync(model, cancellationToken: cancellationToken).ConfigureAwait(false); |
| | | 120 | | } |
| | 3 | 121 | | } |
| | | 122 | | |
| | | 123 | | public async Task SaveRecoveryStateAsync(string correlationId, RecoveryState state, TimeSpan ttl, CancellationToken |
| | | 124 | | { |
| | 1 | 125 | | await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); |
| | | 126 | | // Upsert pipeline stamped with the server clock ($$NOW), matching the message-side |
| | | 127 | | // discipline (and the PG/SqlServer DB-clock discipline): app-clock expiry math would shift |
| | | 128 | | // the recovery window by whatever the client clock is skewed. |
| | 1 | 129 | | await _recovery.UpdateOneAsync( |
| | 1 | 130 | | Builders<MongoRecoveryStateDocument>.Filter.Eq(item => item.Id, RegistrationKey(correlationId, state.Registr |
| | 1 | 131 | | BuildRecoveryStateUpsertPipeline(correlationId, state, ttl), |
| | 1 | 132 | | new UpdateOptions { IsUpsert = true }, |
| | 1 | 133 | | cancellationToken).ConfigureAwait(false); |
| | 1 | 134 | | } |
| | | 135 | | |
| | | 136 | | /// <summary> |
| | | 137 | | /// Upsert pipeline for a recovery registration: every field is overwritten (a re-save refreshes |
| | | 138 | | /// the registration), with <c>expires_at</c>/<c>registered_at</c> computed on the server clock. |
| | | 139 | | /// </summary> |
| | | 140 | | internal static UpdateDefinition<MongoRecoveryStateDocument> BuildRecoveryStateUpsertPipeline( |
| | | 141 | | string correlationId, |
| | | 142 | | RecoveryState state, |
| | | 143 | | TimeSpan ttl) |
| | 1 | 144 | | => Builders<MongoRecoveryStateDocument>.Update.Pipeline(new[] |
| | 1 | 145 | | { |
| | 1 | 146 | | new BsonDocument("$set", new BsonDocument |
| | 1 | 147 | | { |
| | 1 | 148 | | ["correlation_id"] = correlationId, |
| | 1 | 149 | | ["registration_id"] = new BsonBinaryData(state.RegistrationId, GuidRepresentation.Standard), |
| | 1 | 150 | | ["state_json"] = AsyncResponseJson.Serialize(state), |
| | 1 | 151 | | ["expires_at"] = new BsonDocument("$add", new BsonArray { "$$NOW", ttl.TotalMilliseconds }), |
| | 1 | 152 | | ["registered_at"] = "$$NOW" |
| | 1 | 153 | | }) |
| | 1 | 154 | | }); |
| | | 155 | | |
| | | 156 | | public async Task<IReadOnlyList<string>> LoadRecoveryStatesAsync(string correlationId, CancellationToken cancellatio |
| | | 157 | | { |
| | 1 | 158 | | await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); |
| | 1 | 159 | | var filter = Builders<MongoRecoveryStateDocument>.Filter.Eq(item => item.CorrelationId, correlationId) |
| | 1 | 160 | | & NotExpiredOnServerClock<MongoRecoveryStateDocument>(); |
| | 1 | 161 | | var documents = await _recovery.Find(filter) |
| | 1 | 162 | | .SortBy(item => item.RegisteredAtUtc) |
| | 1 | 163 | | .Project(item => item.StateJson) |
| | 1 | 164 | | .ToListAsync(cancellationToken).ConfigureAwait(false); |
| | 1 | 165 | | return documents; |
| | 1 | 166 | | } |
| | | 167 | | |
| | | 168 | | public async Task<bool> DeleteRecoveryStateAsync(string correlationId, Guid registrationId, CancellationToken cancel |
| | | 169 | | { |
| | 1 | 170 | | await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); |
| | 1 | 171 | | var single = await _recovery.DeleteOneAsync( |
| | 1 | 172 | | Builders<MongoRecoveryStateDocument>.Filter.Eq(item => item.Id, RegistrationKey(correlationId, registrationI |
| | 1 | 173 | | cancellationToken).ConfigureAwait(false); |
| | 1 | 174 | | return single.DeletedCount > 0; |
| | 1 | 175 | | } |
| | | 176 | | |
| | | 177 | | public async IAsyncEnumerable<string> ScanRecoveryStateJsonAsync([EnumeratorCancellation] CancellationToken cancella |
| | | 178 | | { |
| | 1 | 179 | | await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); |
| | 1 | 180 | | var filter = NotExpiredOnServerClock<MongoRecoveryStateDocument>(); |
| | 1 | 181 | | using var cursor = await _recovery.Find(filter) |
| | 1 | 182 | | .SortBy(item => item.RegisteredAtUtc) |
| | 1 | 183 | | .Project(item => item.StateJson) |
| | 1 | 184 | | .ToCursorAsync(cancellationToken).ConfigureAwait(false); |
| | 1 | 185 | | while (await cursor.MoveNextAsync(cancellationToken).ConfigureAwait(false)) |
| | | 186 | | { |
| | 1 | 187 | | foreach (var json in cursor.Current) |
| | 1 | 188 | | yield return json; |
| | | 189 | | } |
| | 1 | 190 | | } |
| | | 191 | | |
| | | 192 | | /// <summary> |
| | | 193 | | /// Inserts a response envelope document. The caller supplies the message id and the write is an |
| | | 194 | | /// upsert that preserves an existing document, so a retried publish is idempotent rather than |
| | | 195 | | /// duplicating the response. Timestamps are stamped with the server clock (<c>$$NOW</c>) so |
| | | 196 | | /// dispatch watermarks never mix client and server clocks. The insert itself is the wake signal: |
| | | 197 | | /// every process's change stream observes it. |
| | | 198 | | /// </summary> |
| | | 199 | | public Task<DateTimeOffset> InsertMessageAsync(Guid id, string correlationId, string envelopeJson, TimeSpan retentio |
| | 1 | 200 | | => AsyncResponseRetry.ExecuteAsync( |
| | 1 | 201 | | token => InsertMessageOnceAsync(id, correlationId, envelopeJson, retention, token), |
| | 1 | 202 | | IsTransient, |
| | 1 | 203 | | _options.PublishMaxAttempts, |
| | 1 | 204 | | _options.PublishRetryBaseDelay, |
| | 1 | 205 | | _options.PublishRetryMaxDelay, |
| | 1 | 206 | | cancellationToken); |
| | | 207 | | |
| | | 208 | | private async Task<DateTimeOffset> InsertMessageOnceAsync(Guid id, string correlationId, string envelopeJson, TimeSp |
| | | 209 | | { |
| | 1 | 210 | | await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); |
| | | 211 | | |
| | | 212 | | // findOneAndUpdate instead of updateOne so the returned document carries the |
| | | 213 | | // server-stamped ($$NOW) created_at — the original document's on a publish retry, per the |
| | | 214 | | // pipeline's $ifNull — for the same-process fast path's watermark comparison. |
| | 1 | 215 | | var document = await _messages.FindOneAndUpdateAsync( |
| | 1 | 216 | | Builders<MongoChannelMessageDocument>.Filter.Eq(item => item.Id, id), |
| | 1 | 217 | | BuildInsertMessagePipeline(correlationId, envelopeJson, retention), |
| | 1 | 218 | | new FindOneAndUpdateOptions<MongoChannelMessageDocument> |
| | 1 | 219 | | { |
| | 1 | 220 | | IsUpsert = true, |
| | 1 | 221 | | ReturnDocument = ReturnDocument.After |
| | 1 | 222 | | }, |
| | 1 | 223 | | cancellationToken).ConfigureAwait(false); |
| | | 224 | | |
| | | 225 | | // Upsert + ReturnDocument.After cannot return null from a healthy server. If a driver |
| | | 226 | | // anomaly ever surfaces one, persistence is UNKNOWN — reporting success with a fabricated |
| | | 227 | | // app-clock timestamp would both lie about it and feed a client clock into the |
| | | 228 | | // server-clock watermark. Fail instead, so the retry/error path runs. |
| | 1 | 229 | | return document is null |
| | 1 | 230 | | ? throw new InvalidOperationException( |
| | 1 | 231 | | $"MongoDB response upsert for message {id} returned no document despite IsUpsert + ReturnDocument.After; |
| | 1 | 232 | | : new DateTimeOffset(document.CreatedAtUtc, TimeSpan.Zero); |
| | 1 | 233 | | } |
| | | 234 | | |
| | | 235 | | /// <summary> |
| | | 236 | | /// Upsert pipeline for a response envelope: <c>$ifNull</c> keeps the original server-stamped |
| | | 237 | | /// timestamps and claim flags when a publish retry finds the document already present. |
| | | 238 | | /// </summary> |
| | | 239 | | internal static UpdateDefinition<MongoChannelMessageDocument> BuildInsertMessagePipeline( |
| | | 240 | | string correlationId, |
| | | 241 | | string envelopeJson, |
| | | 242 | | TimeSpan retention) |
| | 3 | 243 | | => Builders<MongoChannelMessageDocument>.Update.Pipeline(new[] |
| | 3 | 244 | | { |
| | 3 | 245 | | new BsonDocument("$set", new BsonDocument |
| | 3 | 246 | | { |
| | 3 | 247 | | ["correlation_id"] = correlationId, |
| | 3 | 248 | | ["envelope_json"] = envelopeJson, |
| | 3 | 249 | | ["created_at"] = new BsonDocument("$ifNull", new BsonArray { "$created_at", "$$NOW" }), |
| | 3 | 250 | | ["expires_at"] = new BsonDocument("$ifNull", new BsonArray |
| | 3 | 251 | | { |
| | 3 | 252 | | "$expires_at", |
| | 3 | 253 | | new BsonDocument("$add", new BsonArray { "$$NOW", retention.TotalMilliseconds }) |
| | 3 | 254 | | }), |
| | 3 | 255 | | ["acked_at"] = new BsonDocument("$ifNull", new BsonArray { "$acked_at", BsonNull.Value }), |
| | 3 | 256 | | ["recovery_claimed"] = new BsonDocument("$ifNull", new BsonArray { "$recovery_claimed", false }) |
| | 3 | 257 | | }) |
| | 3 | 258 | | }); |
| | | 259 | | |
| | | 260 | | public async Task<IReadOnlyList<MongoDbChannelMessage>> LoadMessagesAsync( |
| | | 261 | | string correlationId, |
| | | 262 | | DateTimeOffset sinceUtc, |
| | | 263 | | int batchSize, |
| | | 264 | | DateTimeOffset? afterCreatedAtUtc, |
| | | 265 | | Guid? afterId, |
| | | 266 | | CancellationToken cancellationToken) |
| | | 267 | | { |
| | 3 | 268 | | await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); |
| | 3 | 269 | | var filter = Builders<MongoChannelMessageDocument>.Filter.Eq(item => item.CorrelationId, correlationId) |
| | 3 | 270 | | & Builders<MongoChannelMessageDocument>.Filter.Gte(item => item.CreatedAtUtc, sinceUtc.UtcDateTime) |
| | 3 | 271 | | & Builders<MongoChannelMessageDocument>.Filter.Gt(item => item.ExpiresAtUtc, DateTime.UtcNow); |
| | 3 | 272 | | if (afterCreatedAtUtc is not null) |
| | | 273 | | { |
| | 3 | 274 | | var afterCreated = afterCreatedAtUtc.Value.UtcDateTime; |
| | 3 | 275 | | var cursorId = afterId ?? throw new ArgumentNullException(nameof(afterId)); |
| | 3 | 276 | | filter &= Builders<MongoChannelMessageDocument>.Filter.Or( |
| | 3 | 277 | | Builders<MongoChannelMessageDocument>.Filter.Gt(item => item.CreatedAtUtc, afterCreated), |
| | 3 | 278 | | Builders<MongoChannelMessageDocument>.Filter.And( |
| | 3 | 279 | | Builders<MongoChannelMessageDocument>.Filter.Eq(item => item.CreatedAtUtc, afterCreated), |
| | 3 | 280 | | Builders<MongoChannelMessageDocument>.Filter.Gt(item => item.Id, cursorId))); |
| | | 281 | | } |
| | 3 | 282 | | var documents = await _messages.Find(filter) |
| | 3 | 283 | | .Sort(Builders<MongoChannelMessageDocument>.Sort |
| | 3 | 284 | | .Ascending(item => item.CreatedAtUtc) |
| | 3 | 285 | | .Ascending(item => item.Id)) |
| | 3 | 286 | | .Limit(batchSize) |
| | 3 | 287 | | .ToListAsync(cancellationToken).ConfigureAwait(false); |
| | 3 | 288 | | var messages = new List<MongoDbChannelMessage>(documents.Count); |
| | 3 | 289 | | foreach (var document in documents) |
| | 3 | 290 | | messages.Add(new MongoDbChannelMessage( |
| | 3 | 291 | | document.Id, |
| | 3 | 292 | | document.CorrelationId, |
| | 3 | 293 | | document.EnvelopeJson, |
| | 3 | 294 | | new DateTimeOffset(document.CreatedAtUtc, TimeSpan.Zero), |
| | 3 | 295 | | document.AckedAtUtc is { } acked ? new DateTimeOffset(acked, TimeSpan.Zero) : null)); |
| | 3 | 296 | | return messages; |
| | 3 | 297 | | } |
| | | 298 | | |
| | | 299 | | /// <summary> |
| | | 300 | | /// Atomically claims a message for live delivery via <c>findOneAndUpdate</c>: sets |
| | | 301 | | /// <c>acked_at</c> unless the publisher has already routed it to the lost-subscriber path |
| | | 302 | | /// (<c>recovery_claimed</c>). Returns <c>false</c> when recovery owns the message, so a |
| | | 303 | | /// slow-but-live waiter does not deliver a response the recovery callback already handled. |
| | | 304 | | /// Multiple processes may each win this claim, preserving cross-process fan-out, because it |
| | | 305 | | /// gates only on <c>recovery_claimed</c>, not on <c>acked_at</c>. |
| | | 306 | | /// </summary> |
| | | 307 | | public async Task<bool> TryClaimForDeliveryAsync(Guid messageId, CancellationToken cancellationToken) |
| | | 308 | | { |
| | 3 | 309 | | await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); |
| | 3 | 310 | | var claimed = await _messages.FindOneAndUpdateAsync( |
| | 3 | 311 | | BuildDeliveryClaimFilter(messageId), |
| | 3 | 312 | | BuildDeliveryClaimUpdate(), |
| | 3 | 313 | | new FindOneAndUpdateOptions<MongoChannelMessageDocument> { ReturnDocument = ReturnDocument.After }, |
| | 3 | 314 | | cancellationToken).ConfigureAwait(false); |
| | 3 | 315 | | return claimed is not null; |
| | 3 | 316 | | } |
| | | 317 | | |
| | | 318 | | internal static FilterDefinition<MongoChannelMessageDocument> BuildDeliveryClaimFilter(Guid messageId) |
| | 3 | 319 | | => Builders<MongoChannelMessageDocument>.Filter.Eq(item => item.Id, messageId) |
| | 3 | 320 | | & Builders<MongoChannelMessageDocument>.Filter.Eq(item => item.RecoveryClaimed, false) |
| | 3 | 321 | | & Builders<MongoChannelMessageDocument>.Filter.Gt(item => item.ExpiresAtUtc, DateTime.UtcNow); |
| | | 322 | | |
| | | 323 | | internal static UpdateDefinition<MongoChannelMessageDocument> BuildDeliveryClaimUpdate() |
| | 3 | 324 | | => Builders<MongoChannelMessageDocument>.Update.Pipeline(new[] |
| | 3 | 325 | | { |
| | 3 | 326 | | new BsonDocument("$set", new BsonDocument( |
| | 3 | 327 | | "acked_at", |
| | 3 | 328 | | new BsonDocument("$ifNull", new BsonArray { "$acked_at", "$$NOW" }))) |
| | 3 | 329 | | }); |
| | | 330 | | |
| | | 331 | | /// <summary> |
| | | 332 | | /// Atomically claims a message for the lost-subscriber path: sets <c>recovery_claimed</c> only |
| | | 333 | | /// while no waiter has delivered (<c>acked_at</c> is still null). Returns <c>true</c> when |
| | | 334 | | /// recovery wins; <c>false</c> means a live waiter already took the message, so the publisher |
| | | 335 | | /// must not also fire the recovery callback. Document-level atomicity of |
| | | 336 | | /// <c>findOneAndUpdate</c> serializes this against <see cref="TryClaimForDeliveryAsync"/>. |
| | | 337 | | /// </summary> |
| | | 338 | | public async Task<bool> TryClaimForRecoveryAsync(Guid messageId, CancellationToken cancellationToken) |
| | | 339 | | { |
| | 3 | 340 | | await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); |
| | 3 | 341 | | var claimed = await _messages.FindOneAndUpdateAsync( |
| | 3 | 342 | | BuildRecoveryClaimFilter(messageId), |
| | 3 | 343 | | Builders<MongoChannelMessageDocument>.Update.Set(item => item.RecoveryClaimed, true), |
| | 3 | 344 | | new FindOneAndUpdateOptions<MongoChannelMessageDocument> { ReturnDocument = ReturnDocument.After }, |
| | 3 | 345 | | cancellationToken).ConfigureAwait(false); |
| | 3 | 346 | | return claimed is not null; |
| | 3 | 347 | | } |
| | | 348 | | |
| | | 349 | | internal static FilterDefinition<MongoChannelMessageDocument> BuildRecoveryClaimFilter(Guid messageId) |
| | 3 | 350 | | => Builders<MongoChannelMessageDocument>.Filter.Eq(item => item.Id, messageId) |
| | 3 | 351 | | & Builders<MongoChannelMessageDocument>.Filter.Eq(item => item.AckedAtUtc, null); |
| | | 352 | | |
| | | 353 | | /// <summary>Returns the server's current UTC time, used as a clock-safe delivery watermark.</summary> |
| | | 354 | | public async Task<DateTimeOffset> GetServerTimeUtcAsync(CancellationToken cancellationToken) |
| | | 355 | | { |
| | 3 | 356 | | var reply = await _database.RunCommandAsync<BsonDocument>( |
| | 3 | 357 | | new BsonDocument("hello", 1), |
| | 3 | 358 | | cancellationToken: cancellationToken).ConfigureAwait(false); |
| | 3 | 359 | | return reply.TryGetValue("localTime", out var localTime) && localTime is BsonDateTime serverTime |
| | 3 | 360 | | ? new DateTimeOffset(serverTime.ToUniversalTime(), TimeSpan.Zero) |
| | 3 | 361 | | : DateTimeOffset.UtcNow; |
| | 3 | 362 | | } |
| | | 363 | | |
| | | 364 | | public async Task<bool> IsMessageAcknowledgedAsync(Guid messageId, CancellationToken cancellationToken) |
| | | 365 | | { |
| | 3 | 366 | | await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); |
| | 3 | 367 | | var filter = Builders<MongoChannelMessageDocument>.Filter.Eq(item => item.Id, messageId) |
| | 3 | 368 | | & Builders<MongoChannelMessageDocument>.Filter.Gt(item => item.ExpiresAtUtc, DateTime.UtcNow); |
| | | 369 | | // Direct member projection, not an anonymous type: anonymous projections lower to the |
| | | 370 | | // RequiresUnreferencedCode Expression.New(ctor, args, members) overload, which the ILC |
| | | 371 | | // trim analysis rejects (Roslyn's analyzer skips compiler-lowered expression trees, so |
| | | 372 | | // only Native AOT publishes catch it). Missing document and unacked document both come |
| | | 373 | | // back as null, which is exactly the contract here. |
| | 3 | 374 | | var ackedAtUtc = await _messages.Find(filter) |
| | 3 | 375 | | .Project(item => item.AckedAtUtc) |
| | 3 | 376 | | .FirstOrDefaultAsync(cancellationToken).ConfigureAwait(false); |
| | 3 | 377 | | return ackedAtUtc is not null; |
| | 3 | 378 | | } |
| | | 379 | | |
| | | 380 | | public async Task UpsertSubscriberAsync(string correlationId, Guid registrationId, string instanceId, TimeSpan ttl, |
| | | 381 | | { |
| | 3 | 382 | | await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); |
| | 3 | 383 | | await _subscribers.UpdateOneAsync( |
| | 3 | 384 | | Builders<MongoChannelSubscriberDocument>.Filter.Eq(item => item.Id, RegistrationKey(correlationId, registrat |
| | 3 | 385 | | BuildSubscriberUpsertPipeline(correlationId, registrationId, instanceId, ttl), |
| | 3 | 386 | | new UpdateOptions { IsUpsert = true }, |
| | 3 | 387 | | cancellationToken).ConfigureAwait(false); |
| | 3 | 388 | | } |
| | | 389 | | |
| | | 390 | | /// <summary> |
| | | 391 | | /// Upsert pipeline for a subscriber liveness document, with <c>expires_at</c> computed on the |
| | | 392 | | /// server clock ($$NOW) — app-clock liveness math would let a skewed client look dead (or |
| | | 393 | | /// immortal) to publishers comparing against server-side expiry. |
| | | 394 | | /// </summary> |
| | | 395 | | internal static UpdateDefinition<MongoChannelSubscriberDocument> BuildSubscriberUpsertPipeline( |
| | | 396 | | string correlationId, |
| | | 397 | | Guid registrationId, |
| | | 398 | | string instanceId, |
| | | 399 | | TimeSpan ttl) |
| | 3 | 400 | | => Builders<MongoChannelSubscriberDocument>.Update.Pipeline(new[] |
| | 3 | 401 | | { |
| | 3 | 402 | | new BsonDocument("$set", new BsonDocument |
| | 3 | 403 | | { |
| | 3 | 404 | | ["correlation_id"] = correlationId, |
| | 3 | 405 | | ["registration_id"] = new BsonBinaryData(registrationId, GuidRepresentation.Standard), |
| | 3 | 406 | | ["instance_id"] = instanceId, |
| | 3 | 407 | | ["expires_at"] = new BsonDocument("$add", new BsonArray { "$$NOW", ttl.TotalMilliseconds }) |
| | 3 | 408 | | }) |
| | 3 | 409 | | }); |
| | | 410 | | |
| | | 411 | | public async Task HeartbeatSubscribersAsync( |
| | | 412 | | string instanceId, |
| | | 413 | | IReadOnlyCollection<(string CorrelationId, Guid RegistrationId)> registrations, |
| | | 414 | | TimeSpan ttl, |
| | | 415 | | CancellationToken cancellationToken) |
| | | 416 | | { |
| | 3 | 417 | | if (registrations.Count == 0) |
| | 3 | 418 | | return; |
| | | 419 | | |
| | 3 | 420 | | await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); |
| | | 421 | | |
| | | 422 | | // Per-registration upserts rather than one bare UpdateMany: the caller only heartbeats |
| | | 423 | | // registrations that are live in this process, so a missing document means the TTL reaper |
| | | 424 | | // deleted it (e.g. after a >timeout stall) — re-creating it here is what brings the waiter |
| | | 425 | | // back from "permanently invisible". Same document shape as UpsertSubscriberAsync. |
| | 3 | 426 | | var writes = new List<WriteModel<MongoChannelSubscriberDocument>>(registrations.Count); |
| | 3 | 427 | | foreach (var (correlationId, registrationId) in registrations) |
| | | 428 | | { |
| | 3 | 429 | | writes.Add(new UpdateOneModel<MongoChannelSubscriberDocument>( |
| | 3 | 430 | | Builders<MongoChannelSubscriberDocument>.Filter.Eq(item => item.Id, RegistrationKey(correlationId, regis |
| | 3 | 431 | | BuildSubscriberUpsertPipeline(correlationId, registrationId, instanceId, ttl)) |
| | 3 | 432 | | { |
| | 3 | 433 | | IsUpsert = true |
| | 3 | 434 | | }); |
| | | 435 | | } |
| | | 436 | | |
| | 3 | 437 | | await _subscribers.BulkWriteAsync( |
| | 3 | 438 | | writes, |
| | 3 | 439 | | new BulkWriteOptions { IsOrdered = false }, |
| | 3 | 440 | | cancellationToken).ConfigureAwait(false); |
| | 3 | 441 | | } |
| | | 442 | | |
| | | 443 | | public async Task DeleteSubscriberAsync(string correlationId, Guid registrationId, CancellationToken cancellationTok |
| | | 444 | | { |
| | 3 | 445 | | await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); |
| | 3 | 446 | | await _subscribers.DeleteOneAsync( |
| | 3 | 447 | | Builders<MongoChannelSubscriberDocument>.Filter.Eq(item => item.Id, RegistrationKey(correlationId, registrat |
| | 3 | 448 | | cancellationToken).ConfigureAwait(false); |
| | 3 | 449 | | } |
| | | 450 | | |
| | | 451 | | public async Task<long> CountActiveSubscribersAsync(string correlationId, CancellationToken cancellationToken) |
| | | 452 | | { |
| | 3 | 453 | | await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); |
| | 3 | 454 | | var filter = Builders<MongoChannelSubscriberDocument>.Filter.Eq(item => item.CorrelationId, correlationId) |
| | 3 | 455 | | & NotExpiredOnServerClock<MongoChannelSubscriberDocument>(); |
| | 3 | 456 | | return await _subscribers.CountDocumentsAsync(filter, cancellationToken: cancellationToken).ConfigureAwait(false |
| | 3 | 457 | | } |
| | | 458 | | |
| | | 459 | | /// <summary> |
| | | 460 | | /// Server-clock expiry filter (<c>$expr: expires_at > $$NOW</c>): liveness and recovery |
| | | 461 | | /// expiry are stamped with $$NOW, so comparing them against the app clock would reintroduce |
| | | 462 | | /// the clock-skew hole the server-side stamps exist to close. |
| | | 463 | | /// </summary> |
| | | 464 | | internal static FilterDefinition<TDocument> NotExpiredOnServerClock<TDocument>() |
| | 3 | 465 | | => new BsonDocument("$expr", new BsonDocument("$gt", new BsonArray { "$expires_at", "$$NOW" })); |
| | | 466 | | |
| | | 467 | | /// <summary> |
| | | 468 | | /// Watches the message collection with a change stream and invokes |
| | | 469 | | /// <paramref name="onNotification"/> with the correlation id of every inserted response. One |
| | | 470 | | /// stream serves every local waiter — the caller routes the id to the right subscription — so |
| | | 471 | | /// waiter count never multiplies server-side cursors. Runs until cancellation or a stream error. |
| | | 472 | | /// </summary> |
| | | 473 | | public async Task WatchMessagesAsync(Func<string?, Task> onNotification, CancellationToken cancellationToken) |
| | | 474 | | { |
| | 3 | 475 | | await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); |
| | 3 | 476 | | using var cursor = await _messages.WatchAsync( |
| | 3 | 477 | | BuildMessageWatchPipeline(), |
| | 3 | 478 | | new ChangeStreamOptions { FullDocument = ChangeStreamFullDocumentOption.UpdateLookup }, |
| | 3 | 479 | | cancellationToken).ConfigureAwait(false); |
| | 3 | 480 | | while (await cursor.MoveNextAsync(cancellationToken).ConfigureAwait(false)) |
| | | 481 | | { |
| | 1 | 482 | | foreach (var change in cursor.Current) |
| | 1 | 483 | | await onNotification(change.FullDocument?.CorrelationId).ConfigureAwait(false); |
| | | 484 | | } |
| | 3 | 485 | | } |
| | | 486 | | |
| | | 487 | | /// <summary> |
| | | 488 | | /// Change-stream pipeline for response wakes: a <c>$match</c> on insert events. The correlation |
| | | 489 | | /// id travels in the event's full document, letting the dispatcher scan only the signaled |
| | | 490 | | /// correlation id — the targeted-wake contract. |
| | | 491 | | /// </summary> |
| | | 492 | | internal static PipelineDefinition<ChangeStreamDocument<MongoChannelMessageDocument>, ChangeStreamDocument<MongoChan |
| | 3 | 493 | | => new EmptyPipelineDefinition<ChangeStreamDocument<MongoChannelMessageDocument>>() |
| | 3 | 494 | | .Match(change => change.OperationType == ChangeStreamOperationType.Insert); |
| | | 495 | | |
| | | 496 | | /// <summary>Returns <c>true</c> when the server rejected the change stream itself (not a transient cursor error).</ |
| | | 497 | | internal static bool IsChangeStreamUnsupported(Exception exception) |
| | 3 | 498 | | => exception is MongoCommandException commandException |
| | 3 | 499 | | && (commandException.Code == 40573 |
| | 3 | 500 | | || commandException.Message.Contains("only supported on replica sets", StringComparison.OrdinalIgnoreCase |
| | | 501 | | |
| | | 502 | | internal static string RegistrationKey(string correlationId, Guid registrationId) |
| | 3 | 503 | | => $"{correlationId}:{registrationId:N}"; |
| | | 504 | | |
| | | 505 | | internal static bool IsTransient(Exception exception) |
| | 3 | 506 | | => exception is not OperationCanceledException |
| | 3 | 507 | | && (exception is MongoConnectionException |
| | 3 | 508 | | or MongoNotPrimaryException |
| | 3 | 509 | | or MongoNodeIsRecoveringException |
| | 3 | 510 | | or MongoExecutionTimeoutException |
| | 3 | 511 | | or TimeoutException |
| | 3 | 512 | | || (exception is MongoException mongoException && mongoException.HasErrorLabel("RetryableWriteError"))); |
| | | 513 | | |
| | | 514 | | /// <summary>Validates a MongoDB collection name coming from options.</summary> |
| | | 515 | | public static void ValidateCollectionName(string? value, string name) |
| | | 516 | | { |
| | 3 | 517 | | if (string.IsNullOrWhiteSpace(value)) |
| | 3 | 518 | | throw new InvalidOperationException($"{nameof(MongoDbAsyncResponseChannelOptions)}.{name} must be configured |
| | 3 | 519 | | if (value.Contains('$') || value.Contains('\0') || value.StartsWith("system.", StringComparison.Ordinal)) |
| | 3 | 520 | | throw new InvalidOperationException( |
| | 3 | 521 | | $"{nameof(MongoDbAsyncResponseChannelOptions)}.{name} '{value}' must be a valid MongoDB collection name |
| | 3 | 522 | | } |
| | | 523 | | |
| | | 524 | | /// <summary>Disposes the Mongo client when the store created (and therefore owns) it.</summary> |
| | | 525 | | public void Dispose() |
| | | 526 | | { |
| | 1 | 527 | | _ensureGate.Dispose(); |
| | 1 | 528 | | (_ownedClient as IDisposable)?.Dispose(); |
| | 0 | 529 | | } |
| | | 530 | | } |
| | | 531 | | |
| | | 532 | | internal sealed class MongoRecoveryStateDocument |
| | | 533 | | { |
| | | 534 | | [BsonId] |
| | | 535 | | [BsonElement("_id")] |
| | | 536 | | public string Id { get; set; } = ""; |
| | | 537 | | |
| | | 538 | | [BsonElement("correlation_id")] |
| | | 539 | | public string CorrelationId { get; set; } = ""; |
| | | 540 | | |
| | | 541 | | [BsonElement("registration_id")] |
| | | 542 | | [BsonGuidRepresentation(GuidRepresentation.Standard)] |
| | | 543 | | public Guid RegistrationId { get; set; } |
| | | 544 | | |
| | | 545 | | [BsonElement("state_json")] |
| | | 546 | | public string StateJson { get; set; } = ""; |
| | | 547 | | |
| | | 548 | | [BsonElement("expires_at")] |
| | | 549 | | public DateTime ExpiresAtUtc { get; set; } |
| | | 550 | | |
| | | 551 | | [BsonElement("registered_at")] |
| | | 552 | | public DateTime RegisteredAtUtc { get; set; } |
| | | 553 | | } |
| | | 554 | | |
| | | 555 | | internal sealed class MongoChannelMessageDocument |
| | | 556 | | { |
| | | 557 | | [BsonId] |
| | | 558 | | [BsonElement("_id")] |
| | | 559 | | [BsonGuidRepresentation(GuidRepresentation.Standard)] |
| | | 560 | | public Guid Id { get; set; } |
| | | 561 | | |
| | | 562 | | [BsonElement("correlation_id")] |
| | | 563 | | public string CorrelationId { get; set; } = ""; |
| | | 564 | | |
| | | 565 | | [BsonElement("envelope_json")] |
| | | 566 | | public string EnvelopeJson { get; set; } = ""; |
| | | 567 | | |
| | | 568 | | [BsonElement("created_at")] |
| | | 569 | | public DateTime CreatedAtUtc { get; set; } |
| | | 570 | | |
| | | 571 | | [BsonElement("expires_at")] |
| | | 572 | | public DateTime ExpiresAtUtc { get; set; } |
| | | 573 | | |
| | | 574 | | [BsonElement("acked_at")] |
| | | 575 | | public DateTime? AckedAtUtc { get; set; } |
| | | 576 | | |
| | | 577 | | [BsonElement("recovery_claimed")] |
| | | 578 | | public bool RecoveryClaimed { get; set; } |
| | | 579 | | } |
| | | 580 | | |
| | | 581 | | internal sealed class MongoChannelSubscriberDocument |
| | | 582 | | { |
| | | 583 | | [BsonId] |
| | | 584 | | [BsonElement("_id")] |
| | | 585 | | public string Id { get; set; } = ""; |
| | | 586 | | |
| | | 587 | | [BsonElement("correlation_id")] |
| | | 588 | | public string CorrelationId { get; set; } = ""; |
| | | 589 | | |
| | | 590 | | [BsonElement("registration_id")] |
| | | 591 | | [BsonGuidRepresentation(GuidRepresentation.Standard)] |
| | | 592 | | public Guid RegistrationId { get; set; } |
| | | 593 | | |
| | | 594 | | [BsonElement("instance_id")] |
| | | 595 | | public string InstanceId { get; set; } = ""; |
| | | 596 | | |
| | | 597 | | [BsonElement("expires_at")] |
| | | 598 | | public DateTime ExpiresAtUtc { get; set; } |
| | | 599 | | } |