| | | 1 | | using Microsoft.Extensions.Logging; |
| | | 2 | | using Microsoft.Extensions.Logging.Abstractions; |
| | | 3 | | using Microsoft.Extensions.Options; |
| | | 4 | | using MongoDB.Bson; |
| | | 5 | | using MongoDB.Bson.Serialization.Attributes; |
| | | 6 | | using MongoDB.Driver; |
| | | 7 | | using System.Runtime.CompilerServices; |
| | | 8 | | using AsyncResponse.Internal; |
| | | 9 | | |
| | | 10 | | namespace AsyncResponse.Channels.MongoDB; |
| | | 11 | | |
| | | 12 | | /// <summary>One stored response envelope row/document as the channel store returns it.</summary> |
| | | 13 | | /// <remarks> |
| | | 14 | | /// <c>EnvelopeJson</c> is the stored envelope, or <c>null</c> for a document the dispatch sweep loaded header-only (an |
| | | 15 | | /// already-acknowledged one — see <see cref="MongoDbChannelStore.LoadMessagesAsync"/>); the |
| | | 16 | | /// sweep hydrates the few such documents it still has to deliver through |
| | | 17 | | /// <see cref="MongoDbChannelStore.LoadMessagesByIdAsync"/> before handing them to a waiter. |
| | | 18 | | /// </remarks> |
| | | 19 | | internal readonly record struct MongoDbChannelMessage( |
| | | 20 | | Guid Id, |
| | | 21 | | string CorrelationId, |
| | | 22 | | string? EnvelopeJson, |
| | | 23 | | DateTimeOffset CreatedAtUtc, |
| | | 24 | | DateTimeOffset? AckedAtUtc = null, |
| | | 25 | | long? AckedSeq = null); |
| | | 26 | | |
| | | 27 | | /// <summary>Document adapter for the MongoDB channel collections and change-stream wake.</summary> |
| | | 28 | | internal sealed class MongoDbChannelStore : IDisposable |
| | | 29 | | { |
| | | 30 | | private readonly IMongoCollection<MongoRecoveryStateDocument> _recovery; |
| | | 31 | | private readonly IMongoCollection<MongoChannelMessageDocument> _messages; |
| | | 32 | | private readonly IMongoCollection<MongoChannelSubscriberDocument> _subscribers; |
| | | 33 | | private readonly IMongoCollection<BsonDocument> _counters; |
| | | 34 | | private readonly IMongoDatabase _database; |
| | | 35 | | private readonly MongoDbAsyncResponseChannelOptions _options; |
| | 540 | 36 | | private readonly SemaphoreSlim _ensureGate = new(1, 1); |
| | | 37 | | private readonly IMongoClient? _ownedClient; |
| | | 38 | | private readonly ILogger _logger; |
| | | 39 | | private bool _created; |
| | | 40 | | |
| | 540 | 41 | | public MongoDbChannelStore( |
| | 540 | 42 | | IMongoDatabase database, |
| | 540 | 43 | | IOptions<MongoDbAsyncResponseChannelOptions> options, |
| | 540 | 44 | | IMongoClient? ownedClient = null, |
| | 540 | 45 | | IMongoNamespaceRegistry? namespaceRegistry = null, |
| | 540 | 46 | | ILogger? logger = null) |
| | | 47 | | { |
| | 540 | 48 | | _options = options.Value; |
| | 540 | 49 | | _options.Validate(); |
| | 540 | 50 | | _database = database; |
| | 540 | 51 | | _logger = logger ?? NullLogger.Instance; |
| | | 52 | | |
| | | 53 | | // Cross-component collection ownership (DI-hosted stores only): another AsyncResponse |
| | | 54 | | // component configured onto one of these collections — the derived counters collection |
| | | 55 | | // included — must fail startup in either construction order. |
| | 540 | 56 | | namespaceRegistry?.Claim( |
| | 540 | 57 | | MongoNamespaceRegistry.ClusterKey(database), |
| | 540 | 58 | | database.DatabaseNamespace.DatabaseName, |
| | 540 | 59 | | "MongoDB channel", |
| | 540 | 60 | | [ |
| | 540 | 61 | | (_options.RecoveryStateCollection, nameof(_options.RecoveryStateCollection)), |
| | 540 | 62 | | (_options.MessageCollection, nameof(_options.MessageCollection)), |
| | 540 | 63 | | (_options.SubscriberCollection, nameof(_options.SubscriberCollection)), |
| | 540 | 64 | | (CountersCollectionName(_options.MessageCollection), "derived ack-counter collection"), |
| | 540 | 65 | | ]); |
| | | 66 | | |
| | | 67 | | // Namespace BYTE limits can only be checked here, where the actual database name is |
| | | 68 | | // first known — and the derived counters namespace is 9 bytes longer than the configured |
| | | 69 | | // message collection, so a near-limit configuration passed every static check and failed |
| | | 70 | | // at the first ack-sequence draw. |
| | 538 | 71 | | MongoNamespaceRegistry.ValidateEffectiveNamespace(database, _options.RecoveryStateCollection, nameof(_options.Re |
| | 538 | 72 | | MongoNamespaceRegistry.ValidateEffectiveNamespace(database, _options.MessageCollection, nameof(_options.MessageC |
| | 536 | 73 | | MongoNamespaceRegistry.ValidateEffectiveNamespace(database, _options.SubscriberCollection, nameof(_options.Subsc |
| | 536 | 74 | | MongoNamespaceRegistry.ValidateEffectiveNamespace(database, CountersCollectionName(_options.MessageCollection), |
| | | 75 | | |
| | | 76 | | // Primary reads override whatever the host-registered client/connection string configured: |
| | | 77 | | // a secondaryPreferred connection routed the liveness probe, recovery-state read and message |
| | | 78 | | // load to a lagging secondary, so a publisher racing a fresh registration saw 0 subscribers |
| | | 79 | | // AND 0 recovery states and dropped the response while reporting success. It also keeps |
| | | 80 | | // reads on the same authority whose $$NOW the expiry filters evaluate against (same |
| | | 81 | | // reasoning as the MongoDB durable-flow store's pin). |
| | 534 | 82 | | _recovery = database.GetCollection<MongoRecoveryStateDocument>(_options.RecoveryStateCollection) |
| | 534 | 83 | | .WithReadPreference(ReadPreference.Primary); |
| | 534 | 84 | | _messages = database.GetCollection<MongoChannelMessageDocument>(_options.MessageCollection) |
| | 534 | 85 | | .WithReadPreference(ReadPreference.Primary); |
| | 534 | 86 | | _subscribers = database.GetCollection<MongoChannelSubscriberDocument>(_options.SubscriberCollection) |
| | 534 | 87 | | .WithReadPreference(ReadPreference.Primary); |
| | | 88 | | // The monotonic ack sequence: delivery claims and subscription registrations draw from |
| | | 89 | | // this ONE counter, giving acked_seq and a subscription's start position a total order no |
| | | 90 | | // pair of same-tick timestamps has. Created on first upsert; no index needed (_id only). |
| | 534 | 91 | | _counters = database.GetCollection<BsonDocument>(CountersCollectionName(_options.MessageCollection)) |
| | 534 | 92 | | .WithReadPreference(ReadPreference.Primary); |
| | 534 | 93 | | _ownedClient = ownedClient; |
| | 534 | 94 | | } |
| | | 95 | | |
| | | 96 | | public async Task EnsureCreatedAsync(CancellationToken cancellationToken = default) |
| | | 97 | | { |
| | 10781 | 98 | | if (_created) |
| | 10305 | 99 | | return; |
| | | 100 | | |
| | 476 | 101 | | await _ensureGate.WaitAsync(cancellationToken).ConfigureAwait(false); |
| | | 102 | | try |
| | | 103 | | { |
| | 476 | 104 | | if (_created) |
| | 2 | 105 | | return; |
| | | 106 | | |
| | | 107 | | // Persisted cross-host ownership: the in-container registry cannot see other hosts |
| | | 108 | | // or directly constructed stores, so claim the effective collections here — before |
| | | 109 | | // any index DDL, and INDEPENDENTLY of AutoCreateIndexes (disabling index DDL must |
| | | 110 | | // not disable collision protection) — and fail startup when another component |
| | | 111 | | // already owns one. |
| | 474 | 112 | | if (_options.UseOwnershipLedger) |
| | | 113 | | { |
| | 428 | 114 | | await MongoOwnershipLedger.ClaimAsync( |
| | 428 | 115 | | _database, |
| | 428 | 116 | | "MongoDB channel", |
| | 428 | 117 | | [ |
| | 428 | 118 | | (_options.RecoveryStateCollection, nameof(_options.RecoveryStateCollection)), |
| | 428 | 119 | | (_options.MessageCollection, nameof(_options.MessageCollection)), |
| | 428 | 120 | | (_options.SubscriberCollection, nameof(_options.SubscriberCollection)), |
| | 428 | 121 | | (CountersCollectionName(_options.MessageCollection), "derived ack-counter collection"), |
| | 428 | 122 | | ], |
| | 428 | 123 | | cancellationToken).ConfigureAwait(false); |
| | | 124 | | } |
| | | 125 | | |
| | 473 | 126 | | if (!_options.AutoCreateIndexes) |
| | | 127 | | { |
| | | 128 | | // Manually managed indexes get a one-time read-only check instead of DDL. There |
| | | 129 | | // is no collection shape to verify (documents are schemaless), so the silent |
| | | 130 | | // failure modes are all indexes — above all a missing TTL index, which means |
| | | 131 | | // nothing ever reaps expired documents. Absence is a warning, never a startup |
| | | 132 | | // failure: indexes degrade retention and performance, not correctness, and a |
| | | 133 | | // least-privilege operator may provision them out of band. |
| | 98 | 134 | | await WarnIfManagedIndexesMissingAsync(cancellationToken).ConfigureAwait(false); |
| | 98 | 135 | | _created = true; |
| | 98 | 136 | | return; |
| | | 137 | | } |
| | | 138 | | |
| | | 139 | | // TTL indexes (expireAfterSeconds = 0 on the expiry timestamp) make MongoDB itself reap |
| | | 140 | | // expired documents — no application-side pruning needed. Reads still filter on the |
| | | 141 | | // expiry because the TTL monitor only runs periodically (~60s). |
| | 375 | 142 | | await CreateTtlIndexAsync( |
| | 375 | 143 | | _recovery, |
| | 375 | 144 | | Builders<MongoRecoveryStateDocument>.IndexKeys.Ascending(item => item.ExpiresAtUtc), |
| | 375 | 145 | | $"{_options.RecoveryStateCollection}_expires_idx", |
| | 375 | 146 | | cancellationToken).ConfigureAwait(false); |
| | 375 | 147 | | await _recovery.Indexes.CreateOneAsync( |
| | 375 | 148 | | new CreateIndexModel<MongoRecoveryStateDocument>( |
| | 375 | 149 | | Builders<MongoRecoveryStateDocument>.IndexKeys |
| | 375 | 150 | | .Ascending(item => item.CorrelationId) |
| | 375 | 151 | | .Ascending(item => item.RegisteredAtUtc), |
| | 375 | 152 | | new CreateIndexOptions { Name = $"{_options.RecoveryStateCollection}_correlation_idx" }), |
| | 375 | 153 | | cancellationToken: cancellationToken).ConfigureAwait(false); |
| | | 154 | | |
| | 375 | 155 | | await CreateTtlIndexAsync( |
| | 375 | 156 | | _messages, |
| | 375 | 157 | | Builders<MongoChannelMessageDocument>.IndexKeys.Ascending(item => item.ExpiresAtUtc), |
| | 375 | 158 | | $"{_options.MessageCollection}_expires_idx", |
| | 375 | 159 | | cancellationToken).ConfigureAwait(false); |
| | 375 | 160 | | await _messages.Indexes.CreateOneAsync( |
| | 375 | 161 | | new CreateIndexModel<MongoChannelMessageDocument>( |
| | 375 | 162 | | Builders<MongoChannelMessageDocument>.IndexKeys |
| | 375 | 163 | | .Ascending(item => item.CorrelationId) |
| | 375 | 164 | | .Ascending(item => item.CreatedAtUtc), |
| | 375 | 165 | | new CreateIndexOptions { Name = $"{_options.MessageCollection}_correlation_created_idx" }), |
| | 375 | 166 | | cancellationToken: cancellationToken).ConfigureAwait(false); |
| | | 167 | | |
| | 375 | 168 | | await CreateTtlIndexAsync( |
| | 375 | 169 | | _subscribers, |
| | 375 | 170 | | Builders<MongoChannelSubscriberDocument>.IndexKeys.Ascending(item => item.ExpiresAtUtc), |
| | 375 | 171 | | $"{_options.SubscriberCollection}_expires_idx", |
| | 375 | 172 | | cancellationToken).ConfigureAwait(false); |
| | 375 | 173 | | await _subscribers.Indexes.CreateOneAsync( |
| | 375 | 174 | | new CreateIndexModel<MongoChannelSubscriberDocument>( |
| | 375 | 175 | | Builders<MongoChannelSubscriberDocument>.IndexKeys.Ascending(item => item.CorrelationId), |
| | 375 | 176 | | new CreateIndexOptions { Name = $"{_options.SubscriberCollection}_correlation_idx" }), |
| | 375 | 177 | | cancellationToken: cancellationToken).ConfigureAwait(false); |
| | | 178 | | |
| | 375 | 179 | | _created = true; |
| | 375 | 180 | | } |
| | | 181 | | finally |
| | | 182 | | { |
| | 476 | 183 | | _ensureGate.Release(); |
| | | 184 | | } |
| | 10780 | 185 | | } |
| | | 186 | | |
| | | 187 | | private static async Task CreateTtlIndexAsync<TDocument>( |
| | | 188 | | IMongoCollection<TDocument> collection, |
| | | 189 | | IndexKeysDefinition<TDocument> keys, |
| | | 190 | | string indexName, |
| | | 191 | | CancellationToken cancellationToken) |
| | | 192 | | { |
| | 1125 | 193 | | var model = new CreateIndexModel<TDocument>( |
| | 1125 | 194 | | keys, |
| | 1125 | 195 | | new CreateIndexOptions { Name = indexName, ExpireAfter = TimeSpan.Zero }); |
| | | 196 | | try |
| | | 197 | | { |
| | 1125 | 198 | | await collection.Indexes.CreateOneAsync(model, cancellationToken: cancellationToken).ConfigureAwait(false); |
| | 1121 | 199 | | } |
| | 4 | 200 | | catch (MongoCommandException ex) when (ex.Code is 85 or 86) |
| | | 201 | | { |
| | | 202 | | // IndexOptionsConflict/IndexKeySpecsConflict: an earlier deployment created the |
| | | 203 | | // same-named index with different options. Replace it in place. |
| | | 204 | | try |
| | | 205 | | { |
| | 4 | 206 | | await collection.Indexes.DropOneAsync(indexName, cancellationToken).ConfigureAwait(false); |
| | 2 | 207 | | } |
| | 2 | 208 | | catch (MongoCommandException dropException) when (dropException.Code == 27) |
| | | 209 | | { |
| | | 210 | | // IndexNotFound: a peer host in the same rolling deploy took the same branch and |
| | | 211 | | // dropped it first. Converge on the recreate below (idempotent for an identical spec). |
| | 2 | 212 | | } |
| | | 213 | | |
| | 4 | 214 | | await collection.Indexes.CreateOneAsync(model, cancellationToken: cancellationToken).ConfigureAwait(false); |
| | | 215 | | } |
| | 1125 | 216 | | } |
| | | 217 | | |
| | | 218 | | private async Task WarnIfManagedIndexesMissingAsync(CancellationToken cancellationToken) |
| | | 219 | | { |
| | | 220 | | try |
| | | 221 | | { |
| | 98 | 222 | | await WarnIfCollectionIndexesMissingAsync(_recovery, _options.RecoveryStateCollection, cancellationToken).Co |
| | 4 | 223 | | await WarnIfCollectionIndexesMissingAsync(_messages, _options.MessageCollection, cancellationToken).Configur |
| | 4 | 224 | | await WarnIfCollectionIndexesMissingAsync(_subscribers, _options.SubscriberCollection, cancellationToken).Co |
| | 4 | 225 | | } |
| | 94 | 226 | | catch (Exception ex) when (ex is not OperationCanceledException) |
| | | 227 | | { |
| | | 228 | | // A deployment that cannot even list indexes (no listIndexes privilege, server |
| | | 229 | | // unreachable at first use) must not lose the actual operation to the check: the |
| | | 230 | | // caller's own store call surfaces any real connectivity failure. |
| | 94 | 231 | | _logger.LogDebug(ex, "Skipping index verification for the manually managed MongoDB channel collections; list |
| | 94 | 232 | | } |
| | 98 | 233 | | } |
| | | 234 | | |
| | | 235 | | private async Task WarnIfCollectionIndexesMissingAsync<TDocument>( |
| | | 236 | | IMongoCollection<TDocument> collection, |
| | | 237 | | string collectionName, |
| | | 238 | | CancellationToken cancellationToken) |
| | | 239 | | { |
| | | 240 | | List<BsonDocument> indexes; |
| | | 241 | | try |
| | | 242 | | { |
| | 106 | 243 | | using var cursor = await collection.Indexes.ListAsync(cancellationToken).ConfigureAwait(false); |
| | 10 | 244 | | indexes = await cursor.ToListAsync(cancellationToken).ConfigureAwait(false); |
| | 10 | 245 | | } |
| | 4 | 246 | | catch (MongoCommandException ex) when (ex.Code == 26) |
| | | 247 | | { |
| | | 248 | | // NamespaceNotFound: the collection does not exist yet. MongoDB creates it bare on |
| | | 249 | | // the first write — which, with index DDL disabled, is exactly a collection with no |
| | | 250 | | // TTL index. |
| | 2 | 251 | | indexes = []; |
| | 2 | 252 | | } |
| | | 253 | | |
| | | 254 | | // Matched by KEY, not by name: operators own the naming of manually provisioned indexes. |
| | 20 | 255 | | if (!indexes.Any(index => IndexLeadsOn(index, "expires_at") && index.Contains("expireAfterSeconds"))) |
| | | 256 | | { |
| | 4 | 257 | | _logger.LogWarning( |
| | 4 | 258 | | "MongoDB collection {Database}.{Collection} has no TTL index on 'expires_at' and AutoCreateIndexes is di |
| | 4 | 259 | | "Expired documents are never reaped, so the collection grows without bound. " + |
| | 4 | 260 | | "Create a TTL index (expireAfterSeconds: 0 on 'expires_at') or enable AutoCreateIndexes.", |
| | 4 | 261 | | _database.DatabaseNamespace.DatabaseName, collectionName); |
| | | 262 | | } |
| | | 263 | | |
| | 28 | 264 | | if (!indexes.Any(index => IndexLeadsOn(index, "correlation_id"))) |
| | | 265 | | { |
| | 4 | 266 | | _logger.LogWarning( |
| | 4 | 267 | | "MongoDB collection {Database}.{Collection} has no index leading on 'correlation_id' and AutoCreateIndex |
| | 4 | 268 | | "Correlation-id lookups scan the whole collection — performance only; create the index to restore indexe |
| | 4 | 269 | | _database.DatabaseNamespace.DatabaseName, collectionName); |
| | | 270 | | } |
| | 12 | 271 | | } |
| | | 272 | | |
| | | 273 | | /// <summary> |
| | | 274 | | /// Whether the listed index's FIRST key field is <paramref name="field"/> — what a prefix |
| | | 275 | | /// lookup uses, and (TTL indexes being single-field) what identifies the TTL index. |
| | | 276 | | /// </summary> |
| | | 277 | | private static bool IndexLeadsOn(BsonDocument index, string field) |
| | 24 | 278 | | => index.TryGetValue("key", out var key) |
| | 24 | 279 | | && key is BsonDocument keyDocument |
| | 24 | 280 | | && keyDocument.ElementCount > 0 |
| | 24 | 281 | | && keyDocument.GetElement(0).Name == field; |
| | | 282 | | |
| | | 283 | | public async Task SaveRecoveryStateAsync(string correlationId, RecoveryState state, TimeSpan ttl, CancellationToken |
| | | 284 | | { |
| | 386 | 285 | | await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); |
| | | 286 | | // Upsert pipeline stamped with the server clock ($$NOW), matching the message-side |
| | | 287 | | // discipline (and the PG/SqlServer DB-clock discipline): app-clock expiry math would shift |
| | | 288 | | // the recovery window by whatever the client clock is skewed. |
| | 386 | 289 | | await _recovery.UpdateOneAsync( |
| | 386 | 290 | | Builders<MongoRecoveryStateDocument>.Filter.Eq(item => item.Id, RegistrationKey(correlationId, state.Registr |
| | 386 | 291 | | BuildRecoveryStateUpsertPipeline(correlationId, state, ttl), |
| | 386 | 292 | | new UpdateOptions { IsUpsert = true }, |
| | 386 | 293 | | cancellationToken).ConfigureAwait(false); |
| | 386 | 294 | | } |
| | | 295 | | |
| | | 296 | | /// <summary> |
| | | 297 | | /// Upsert pipeline for a recovery registration: every field is overwritten (a re-save refreshes |
| | | 298 | | /// the registration), with <c>expires_at</c>/<c>registered_at</c> computed on the server clock. |
| | | 299 | | /// </summary> |
| | | 300 | | internal static UpdateDefinition<MongoRecoveryStateDocument> BuildRecoveryStateUpsertPipeline( |
| | | 301 | | string correlationId, |
| | | 302 | | RecoveryState state, |
| | | 303 | | TimeSpan ttl) |
| | 386 | 304 | | => Builders<MongoRecoveryStateDocument>.Update.Pipeline(new[] |
| | 386 | 305 | | { |
| | 386 | 306 | | new BsonDocument("$set", new BsonDocument |
| | 386 | 307 | | { |
| | 386 | 308 | | ["correlation_id"] = correlationId, |
| | 386 | 309 | | ["registration_id"] = new BsonBinaryData(state.RegistrationId, GuidRepresentation.Standard), |
| | 386 | 310 | | ["state_json"] = AsyncResponseJson.Serialize(state), |
| | 386 | 311 | | ["expires_at"] = new BsonDocument("$add", new BsonArray { "$$NOW", ttl.TotalMilliseconds }), |
| | 386 | 312 | | ["registered_at"] = "$$NOW" |
| | 386 | 313 | | }) |
| | 386 | 314 | | }); |
| | | 315 | | |
| | | 316 | | public async Task<IReadOnlyList<string>> LoadRecoveryStatesAsync(string correlationId, CancellationToken cancellatio |
| | | 317 | | { |
| | 33 | 318 | | await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); |
| | 33 | 319 | | var filter = Builders<MongoRecoveryStateDocument>.Filter.Eq(item => item.CorrelationId, correlationId) |
| | 33 | 320 | | & NotExpiredOnServerClock<MongoRecoveryStateDocument>(); |
| | 33 | 321 | | var documents = await _recovery.Find(filter) |
| | 33 | 322 | | .SortBy(item => item.RegisteredAtUtc) |
| | 33 | 323 | | .Project(item => item.StateJson) |
| | 33 | 324 | | .ToListAsync(cancellationToken).ConfigureAwait(false); |
| | 33 | 325 | | return documents; |
| | 33 | 326 | | } |
| | | 327 | | |
| | | 328 | | public async Task<bool> DeleteRecoveryStateAsync(string correlationId, Guid registrationId, CancellationToken cancel |
| | | 329 | | { |
| | 385 | 330 | | await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); |
| | 385 | 331 | | var single = await _recovery.DeleteOneAsync( |
| | 385 | 332 | | Builders<MongoRecoveryStateDocument>.Filter.Eq(item => item.Id, RegistrationKey(correlationId, registrationI |
| | 385 | 333 | | cancellationToken).ConfigureAwait(false); |
| | 385 | 334 | | return single.DeletedCount > 0; |
| | 385 | 335 | | } |
| | | 336 | | |
| | | 337 | | public async IAsyncEnumerable<string> ScanRecoveryStateJsonAsync([EnumeratorCancellation] CancellationToken cancella |
| | | 338 | | { |
| | 1 | 339 | | await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); |
| | 1 | 340 | | var filter = NotExpiredOnServerClock<MongoRecoveryStateDocument>(); |
| | 1 | 341 | | using var cursor = await _recovery.Find(filter) |
| | 1 | 342 | | .SortBy(item => item.RegisteredAtUtc) |
| | 1 | 343 | | .Project(item => item.StateJson) |
| | 1 | 344 | | .ToCursorAsync(cancellationToken).ConfigureAwait(false); |
| | 2 | 345 | | while (await cursor.MoveNextAsync(cancellationToken).ConfigureAwait(false)) |
| | | 346 | | { |
| | 4 | 347 | | foreach (var json in cursor.Current) |
| | 1 | 348 | | yield return json; |
| | | 349 | | } |
| | 1 | 350 | | } |
| | | 351 | | |
| | | 352 | | /// <summary> |
| | | 353 | | /// Inserts a response envelope document. The caller supplies the message id and the write is an |
| | | 354 | | /// upsert that preserves an existing document, so a retried publish is idempotent rather than |
| | | 355 | | /// duplicating the response. Timestamps are stamped with the server clock (<c>$$NOW</c>) so |
| | | 356 | | /// dispatch watermarks never mix client and server clocks. The insert itself is the wake signal: |
| | | 357 | | /// every process's change stream observes it. |
| | | 358 | | /// </summary> |
| | | 359 | | public Task<MongoDbChannelMessage> InsertMessageAsync(Guid id, string correlationId, string envelopeJson, TimeSpan r |
| | 565 | 360 | | => AsyncResponseRetry.ExecuteAsync( |
| | 565 | 361 | | token => InsertMessageOnceAsync(id, correlationId, envelopeJson, retention, token), |
| | 565 | 362 | | MongoTransientFaults.IsTransient, |
| | 565 | 363 | | _options.PublishMaxAttempts, |
| | 565 | 364 | | _options.PublishRetryBaseDelay, |
| | 565 | 365 | | _options.PublishRetryMaxDelay, |
| | 565 | 366 | | cancellationToken); |
| | | 367 | | |
| | | 368 | | private async Task<MongoDbChannelMessage> InsertMessageOnceAsync(Guid id, string correlationId, string envelopeJson, |
| | | 369 | | { |
| | 565 | 370 | | await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); |
| | | 371 | | |
| | | 372 | | // findOneAndUpdate instead of updateOne so the returned document carries the |
| | | 373 | | // server-stamped ($$NOW) created_at — the original document's on a publish retry, per the |
| | | 374 | | // pipeline's $ifNull, together with its settlement columns — for the same-process fast |
| | | 375 | | // path's watermark comparison (a fabricated null acked_at replayed an already-consumed |
| | | 376 | | // response to a waiter registered after the ack). |
| | 565 | 377 | | var document = await _messages.FindOneAndUpdateAsync( |
| | 565 | 378 | | Builders<MongoChannelMessageDocument>.Filter.Eq(item => item.Id, id), |
| | 565 | 379 | | BuildInsertMessagePipeline(correlationId, envelopeJson, retention), |
| | 565 | 380 | | new FindOneAndUpdateOptions<MongoChannelMessageDocument> |
| | 565 | 381 | | { |
| | 565 | 382 | | IsUpsert = true, |
| | 565 | 383 | | ReturnDocument = ReturnDocument.After |
| | 565 | 384 | | }, |
| | 565 | 385 | | cancellationToken).ConfigureAwait(false); |
| | | 386 | | |
| | | 387 | | // Upsert + ReturnDocument.After cannot return null from a healthy server. If a driver |
| | | 388 | | // anomaly ever surfaces one, persistence is UNKNOWN — reporting success with a fabricated |
| | | 389 | | // app-clock timestamp would both lie about it and feed a client clock into the |
| | | 390 | | // server-clock watermark. Fail instead, so the retry/error path runs. |
| | 565 | 391 | | return document is null |
| | 565 | 392 | | ? throw new InvalidOperationException( |
| | 565 | 393 | | $"MongoDB response upsert for message {id} returned no document despite IsUpsert + ReturnDocument.After; |
| | 565 | 394 | | : new MongoDbChannelMessage( |
| | 565 | 395 | | id, |
| | 565 | 396 | | correlationId, |
| | 565 | 397 | | envelopeJson, |
| | 565 | 398 | | new DateTimeOffset(document.CreatedAtUtc, TimeSpan.Zero), |
| | 565 | 399 | | document.AckedAtUtc is { } acked ? new DateTimeOffset(acked, TimeSpan.Zero) : null, |
| | 565 | 400 | | document.AckedSeq); |
| | 565 | 401 | | } |
| | | 402 | | |
| | | 403 | | /// <summary> |
| | | 404 | | /// Upsert pipeline for a response envelope: <c>$ifNull</c> keeps the original server-stamped |
| | | 405 | | /// timestamps and claim flags when a publish retry finds the document already present. |
| | | 406 | | /// </summary> |
| | | 407 | | internal static UpdateDefinition<MongoChannelMessageDocument> BuildInsertMessagePipeline( |
| | | 408 | | string correlationId, |
| | | 409 | | string envelopeJson, |
| | | 410 | | TimeSpan retention) |
| | 567 | 411 | | => Builders<MongoChannelMessageDocument>.Update.Pipeline(new[] |
| | 567 | 412 | | { |
| | 567 | 413 | | new BsonDocument("$set", new BsonDocument |
| | 567 | 414 | | { |
| | 567 | 415 | | ["correlation_id"] = correlationId, |
| | 567 | 416 | | ["envelope_json"] = envelopeJson, |
| | 567 | 417 | | ["created_at"] = new BsonDocument("$ifNull", new BsonArray { "$created_at", "$$NOW" }), |
| | 567 | 418 | | ["expires_at"] = new BsonDocument("$ifNull", new BsonArray |
| | 567 | 419 | | { |
| | 567 | 420 | | "$expires_at", |
| | 567 | 421 | | new BsonDocument("$add", new BsonArray { "$$NOW", retention.TotalMilliseconds }) |
| | 567 | 422 | | }), |
| | 567 | 423 | | ["acked_at"] = new BsonDocument("$ifNull", new BsonArray { "$acked_at", BsonNull.Value }), |
| | 567 | 424 | | ["recovery_claimed"] = new BsonDocument("$ifNull", new BsonArray { "$recovery_claimed", false }) |
| | 567 | 425 | | }) |
| | 567 | 426 | | }); |
| | | 427 | | |
| | | 428 | | public async Task<IReadOnlyList<MongoDbChannelMessage>> LoadMessagesAsync( |
| | | 429 | | string correlationId, |
| | | 430 | | DateTimeOffset sinceUtc, |
| | | 431 | | int batchSize, |
| | | 432 | | DateTimeOffset? afterCreatedAtUtc, |
| | | 433 | | Guid? afterId, |
| | | 434 | | CancellationToken cancellationToken) |
| | | 435 | | { |
| | 1484 | 436 | | await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); |
| | 1484 | 437 | | var filter = Builders<MongoChannelMessageDocument>.Filter.Eq(item => item.CorrelationId, correlationId) |
| | 1484 | 438 | | & Builders<MongoChannelMessageDocument>.Filter.Gte(item => item.CreatedAtUtc, sinceUtc.UtcDateTime) |
| | 1484 | 439 | | & NotExpiredOnServerClock<MongoChannelMessageDocument>(); |
| | 1484 | 440 | | if (afterCreatedAtUtc is not null) |
| | | 441 | | { |
| | 687 | 442 | | var afterCreated = afterCreatedAtUtc.Value.UtcDateTime; |
| | 687 | 443 | | var cursorId = afterId ?? throw new ArgumentNullException(nameof(afterId)); |
| | 687 | 444 | | filter &= Builders<MongoChannelMessageDocument>.Filter.Or( |
| | 687 | 445 | | Builders<MongoChannelMessageDocument>.Filter.Gt(item => item.CreatedAtUtc, afterCreated), |
| | 687 | 446 | | Builders<MongoChannelMessageDocument>.Filter.And( |
| | 687 | 447 | | Builders<MongoChannelMessageDocument>.Filter.Eq(item => item.CreatedAtUtc, afterCreated), |
| | 687 | 448 | | Builders<MongoChannelMessageDocument>.Filter.Gt(item => item.Id, cursorId))); |
| | | 449 | | } |
| | 1484 | 450 | | var documents = await _messages.Find(filter) |
| | 1484 | 451 | | .Project(SweepProjection) |
| | 1484 | 452 | | .Sort(Builders<MongoChannelMessageDocument>.Sort |
| | 1484 | 453 | | .Ascending(item => item.CreatedAtUtc) |
| | 1484 | 454 | | .Ascending(item => item.Id)) |
| | 1484 | 455 | | .Limit(batchSize) |
| | 1484 | 456 | | .ToListAsync(cancellationToken).ConfigureAwait(false); |
| | 1460 | 457 | | return ToMessages(documents); |
| | 1460 | 458 | | } |
| | | 459 | | |
| | | 460 | | /// <summary> |
| | | 461 | | /// The sweep's projection: every field but the envelope, and the envelope only for a document |
| | | 462 | | /// nobody has acknowledged yet. Acknowledged documents are the consumed history the sweep |
| | | 463 | | /// re-reads on every tick (they stay in the result so a fan-out waiter in ANOTHER process |
| | | 464 | | /// still receives them): shipping their bodies with each sweep made a long-lived progress |
| | | 465 | | /// subscription's cost grow with its whole retained history. The shared sweep fetches the |
| | | 466 | | /// envelope by id for the rare acknowledged document a live subscription has not seen. |
| | | 467 | | /// <c>$ifNull</c> folds a missing <c>acked_at</c> (a pre-settlement document) into null. |
| | | 468 | | /// </summary> |
| | 12 | 469 | | internal static readonly ProjectionDefinition<MongoChannelMessageDocument, MongoChannelMessageDocument> SweepProject |
| | 12 | 470 | | new BsonDocumentProjectionDefinition<MongoChannelMessageDocument, MongoChannelMessageDocument>(new BsonDocument |
| | 12 | 471 | | { |
| | 12 | 472 | | ["_id"] = 1, |
| | 12 | 473 | | ["correlation_id"] = 1, |
| | 12 | 474 | | ["created_at"] = 1, |
| | 12 | 475 | | ["expires_at"] = 1, |
| | 12 | 476 | | ["acked_at"] = 1, |
| | 12 | 477 | | ["acked_seq"] = 1, |
| | 12 | 478 | | ["recovery_claimed"] = 1, |
| | 12 | 479 | | ["envelope_json"] = new BsonDocument("$cond", new BsonArray |
| | 12 | 480 | | { |
| | 12 | 481 | | new BsonDocument("$eq", new BsonArray |
| | 12 | 482 | | { |
| | 12 | 483 | | new BsonDocument("$ifNull", new BsonArray { "$acked_at", BsonNull.Value }), |
| | 12 | 484 | | BsonNull.Value |
| | 12 | 485 | | }), |
| | 12 | 486 | | "$envelope_json", |
| | 12 | 487 | | BsonNull.Value |
| | 12 | 488 | | }) |
| | 12 | 489 | | }); |
| | | 490 | | |
| | | 491 | | /// <summary> |
| | | 492 | | /// The full documents (envelope included) for <paramref name="ids"/> under |
| | | 493 | | /// <paramref name="correlationId"/>, in sweep order — how the dispatch sweep hydrates the |
| | | 494 | | /// header-only acknowledged documents it still has to deliver. A document reaped between the |
| | | 495 | | /// sweep's page and this read is simply absent. |
| | | 496 | | /// </summary> |
| | | 497 | | public async Task<IReadOnlyList<MongoDbChannelMessage>> LoadMessagesByIdAsync( |
| | | 498 | | string correlationId, |
| | | 499 | | IReadOnlyList<Guid> ids, |
| | | 500 | | CancellationToken cancellationToken) |
| | | 501 | | { |
| | 65 | 502 | | if (ids.Count == 0) |
| | 0 | 503 | | return []; |
| | | 504 | | |
| | 65 | 505 | | await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); |
| | 65 | 506 | | var filter = Builders<MongoChannelMessageDocument>.Filter.Eq(item => item.CorrelationId, correlationId) |
| | 65 | 507 | | & Builders<MongoChannelMessageDocument>.Filter.In(item => item.Id, ids) |
| | 65 | 508 | | & NotExpiredOnServerClock<MongoChannelMessageDocument>(); |
| | 65 | 509 | | var documents = await _messages.Find(filter) |
| | 65 | 510 | | .Sort(Builders<MongoChannelMessageDocument>.Sort |
| | 65 | 511 | | .Ascending(item => item.CreatedAtUtc) |
| | 65 | 512 | | .Ascending(item => item.Id)) |
| | 65 | 513 | | .ToListAsync(cancellationToken).ConfigureAwait(false); |
| | 65 | 514 | | return ToMessages(documents); |
| | 65 | 515 | | } |
| | | 516 | | |
| | | 517 | | private static List<MongoDbChannelMessage> ToMessages(List<MongoChannelMessageDocument> documents) |
| | | 518 | | { |
| | 1525 | 519 | | var messages = new List<MongoDbChannelMessage>(documents.Count); |
| | 15178 | 520 | | foreach (var document in documents) |
| | 6064 | 521 | | messages.Add(new MongoDbChannelMessage( |
| | 6064 | 522 | | document.Id, |
| | 6064 | 523 | | document.CorrelationId, |
| | 6064 | 524 | | document.EnvelopeJson, |
| | 6064 | 525 | | new DateTimeOffset(document.CreatedAtUtc, TimeSpan.Zero), |
| | 6064 | 526 | | document.AckedAtUtc is { } acked ? new DateTimeOffset(acked, TimeSpan.Zero) : null, |
| | 6064 | 527 | | document.AckedSeq)); |
| | 1525 | 528 | | return messages; |
| | | 529 | | } |
| | | 530 | | |
| | | 531 | | /// <summary> |
| | | 532 | | /// Atomically claims a message for live delivery via <c>findOneAndUpdate</c>: sets |
| | | 533 | | /// <c>acked_at</c> unless the publisher has already routed it to the lost-subscriber path |
| | | 534 | | /// (<c>recovery_claimed</c>). Returns <c>false</c> when recovery owns the message, so a |
| | | 535 | | /// slow-but-live waiter does not deliver a response the recovery callback already handled. |
| | | 536 | | /// Multiple processes may each win this claim, preserving cross-process fan-out, because it |
| | | 537 | | /// gates only on <c>recovery_claimed</c>, not on <c>acked_at</c>. |
| | | 538 | | /// </summary> |
| | | 539 | | public async Task<bool> TryClaimForDeliveryAsync(Guid messageId, CancellationToken cancellationToken) |
| | | 540 | | { |
| | 5205 | 541 | | await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); |
| | | 542 | | // The sequence value is drawn BEFORE the claim lands, so its position reflects when this |
| | | 543 | | // delivery happened relative to subscription registrations (which draw from the same |
| | | 544 | | // counter). An unused draw on an already-acked row leaves a harmless gap; $ifNull keeps |
| | | 545 | | // the FIRST claim's stamp, mirroring acked_at. |
| | 5205 | 546 | | var ackSeq = await DrawAckSequenceAsync(cancellationToken).ConfigureAwait(false); |
| | 5205 | 547 | | var claimed = await _messages.FindOneAndUpdateAsync( |
| | 5205 | 548 | | BuildDeliveryClaimFilter(messageId), |
| | 5205 | 549 | | BuildDeliveryClaimUpdate(ackSeq), |
| | 5205 | 550 | | new FindOneAndUpdateOptions<MongoChannelMessageDocument> { ReturnDocument = ReturnDocument.After }, |
| | 5205 | 551 | | cancellationToken).ConfigureAwait(false); |
| | 5197 | 552 | | return claimed is not null; |
| | 5197 | 553 | | } |
| | | 554 | | |
| | | 555 | | internal static FilterDefinition<MongoChannelMessageDocument> BuildDeliveryClaimFilter(Guid messageId) |
| | 5207 | 556 | | => Builders<MongoChannelMessageDocument>.Filter.Eq(item => item.Id, messageId) |
| | 5207 | 557 | | & Builders<MongoChannelMessageDocument>.Filter.Eq(item => item.RecoveryClaimed, false) |
| | 5207 | 558 | | & NotExpiredOnServerClock<MongoChannelMessageDocument>(); |
| | | 559 | | |
| | | 560 | | internal static UpdateDefinition<MongoChannelMessageDocument> BuildDeliveryClaimUpdate(long ackSeq) |
| | 5207 | 561 | | => Builders<MongoChannelMessageDocument>.Update.Pipeline(new[] |
| | 5207 | 562 | | { |
| | 5207 | 563 | | // Both fields are computed from the PRE-update document (one $set stage), and the |
| | 5207 | 564 | | // sequence is stamped ONLY when this same update transitions acked_at from null: a |
| | 5207 | 565 | | // row acked by a pre-sequence build must stay permanently unsequenced — back-filling |
| | 5207 | 566 | | // it on a later fan-out re-claim would pair an OLD acked_at with a FRESH sequence |
| | 5207 | 567 | | // value, and a waiter that registered in the original ack's tick would then read the |
| | 5207 | 568 | | // tie as post-registration fan-out, replaying a response its predecessor consumed. |
| | 5207 | 569 | | new BsonDocument("$set", new BsonDocument |
| | 5207 | 570 | | { |
| | 5207 | 571 | | ["acked_at"] = new BsonDocument("$ifNull", new BsonArray { "$acked_at", "$$NOW" }), |
| | 5207 | 572 | | ["acked_seq"] = new BsonDocument("$cond", new BsonArray |
| | 5207 | 573 | | { |
| | 5207 | 574 | | new BsonDocument("$eq", new BsonArray { new BsonDocument("$ifNull", new BsonArray { "$acked_at", Bso |
| | 5207 | 575 | | ackSeq, |
| | 5207 | 576 | | "$acked_seq" |
| | 5207 | 577 | | }) |
| | 5207 | 578 | | }) |
| | 5207 | 579 | | }); |
| | | 580 | | |
| | | 581 | | private async Task<long> DrawAckSequenceAsync(CancellationToken cancellationToken) |
| | | 582 | | { |
| | 5205 | 583 | | var counter = await _counters.FindOneAndUpdateAsync<BsonDocument>( |
| | 5205 | 584 | | new BsonDocument("_id", "ack_seq"), |
| | 5205 | 585 | | new BsonDocument("$inc", new BsonDocument("seq", 1L)), |
| | 5205 | 586 | | new FindOneAndUpdateOptions<BsonDocument, BsonDocument> { IsUpsert = true, ReturnDocument = ReturnDocument.A |
| | 5205 | 587 | | cancellationToken).ConfigureAwait(false); |
| | 5205 | 588 | | return counter["seq"].ToInt64(); |
| | 5205 | 589 | | } |
| | | 590 | | |
| | | 591 | | /// <summary> |
| | | 592 | | /// Atomically claims a message for the lost-subscriber path: sets <c>recovery_claimed</c> only |
| | | 593 | | /// while no waiter has delivered (<c>acked_at</c> is still null). Returns <c>true</c> when |
| | | 594 | | /// recovery wins; <c>false</c> means a live waiter already took the message, so the publisher |
| | | 595 | | /// must not also fire the recovery callback. Document-level atomicity of |
| | | 596 | | /// <c>findOneAndUpdate</c> serializes this against <see cref="TryClaimForDeliveryAsync"/>. |
| | | 597 | | /// </summary> |
| | | 598 | | public async Task<bool> TryClaimForRecoveryAsync(Guid messageId, CancellationToken cancellationToken) |
| | | 599 | | { |
| | 12 | 600 | | await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); |
| | 12 | 601 | | var claimed = await _messages.FindOneAndUpdateAsync( |
| | 12 | 602 | | BuildRecoveryClaimFilter(messageId), |
| | 12 | 603 | | Builders<MongoChannelMessageDocument>.Update.Set(item => item.RecoveryClaimed, true), |
| | 12 | 604 | | new FindOneAndUpdateOptions<MongoChannelMessageDocument> { ReturnDocument = ReturnDocument.After }, |
| | 12 | 605 | | cancellationToken).ConfigureAwait(false); |
| | 12 | 606 | | return claimed is not null; |
| | 12 | 607 | | } |
| | | 608 | | |
| | | 609 | | internal static FilterDefinition<MongoChannelMessageDocument> BuildRecoveryClaimFilter(Guid messageId) |
| | 14 | 610 | | => Builders<MongoChannelMessageDocument>.Filter.Eq(item => item.Id, messageId) |
| | 14 | 611 | | & Builders<MongoChannelMessageDocument>.Filter.Eq(item => item.AckedAtUtc, null); |
| | | 612 | | |
| | | 613 | | /// <summary>Returns the server's current UTC time, used as a clock-safe delivery watermark.</summary> |
| | | 614 | | public async Task<DateTimeOffset> GetServerTimeUtcAsync(CancellationToken cancellationToken) |
| | | 615 | | { |
| | | 616 | | // Primary, explicitly: the watermark must come from the clock whose $$NOW stamps the rows |
| | | 617 | | // it bounds — a secondary's localTime can lag or skew from the primary's. |
| | 3 | 618 | | var reply = await _database.RunCommandAsync<BsonDocument>( |
| | 3 | 619 | | new BsonDocument("hello", 1), |
| | 3 | 620 | | ReadPreference.Primary, |
| | 3 | 621 | | cancellationToken).ConfigureAwait(false); |
| | 3 | 622 | | return reply.TryGetValue("localTime", out var localTime) && localTime is BsonDateTime serverTime |
| | 3 | 623 | | ? new DateTimeOffset(serverTime.ToUniversalTime(), TimeSpan.Zero) |
| | 3 | 624 | | : DateTimeOffset.UtcNow; |
| | 3 | 625 | | } |
| | | 626 | | |
| | | 627 | | /// <summary> |
| | | 628 | | /// A subscription's registration watermark: the server's UTC clock (for the created-at bound) |
| | | 629 | | /// and a fresh position in the monotonic ack sequence (for the exact acked-history bound — |
| | | 630 | | /// see the watermark in the shared channel base). Drawn by ONE atomic counter update whose |
| | | 631 | | /// pipeline advances the sequence and stamps <c>$$NOW</c> in the same document write |
| | | 632 | | /// (PG/SqlServer single-statement parity): with separate clock and sequence round trips, a |
| | | 633 | | /// delivery claim landing between them pairs a same-millisecond <c>acked_at</c> with a lower |
| | | 634 | | /// sequence, and the same-tick tie-breaker then files a legitimate fan-out delivery as |
| | | 635 | | /// history. |
| | | 636 | | /// </summary> |
| | | 637 | | public async Task<(DateTimeOffset ServerTimeUtc, long StartSeq)> GetSubscriptionStartAsync(CancellationToken cancell |
| | | 638 | | { |
| | 393 | 639 | | var counter = await _counters.FindOneAndUpdateAsync<BsonDocument>( |
| | 393 | 640 | | new BsonDocument("_id", "ack_seq"), |
| | 393 | 641 | | BuildSubscriptionStartPipeline(), |
| | 393 | 642 | | new FindOneAndUpdateOptions<BsonDocument, BsonDocument> { IsUpsert = true, ReturnDocument = ReturnDocument.A |
| | 393 | 643 | | cancellationToken).ConfigureAwait(false); |
| | | 644 | | |
| | | 645 | | // The pipeline stamps drawn_at from $$NOW unconditionally, so anything else is a driver |
| | | 646 | | // anomaly. Failing beats an app-clock fallback, which would silently feed a client clock |
| | | 647 | | // into the server-clock watermark this draw exists to protect. |
| | 393 | 648 | | return counter is not null |
| | 393 | 649 | | && counter.TryGetValue("drawn_at", out var drawnAt) |
| | 393 | 650 | | && drawnAt is BsonDateTime serverTime |
| | 393 | 651 | | ? (new DateTimeOffset(serverTime.ToUniversalTime(), TimeSpan.Zero), counter["seq"].ToInt64()) |
| | 393 | 652 | | : throw new InvalidOperationException( |
| | 393 | 653 | | "MongoDB subscription-start draw returned no server-stamped counter document despite IsUpsert + ReturnDo |
| | 391 | 654 | | } |
| | | 655 | | |
| | | 656 | | /// <summary> |
| | | 657 | | /// Counter-update pipeline for a subscription registration: advances the monotonic ack |
| | | 658 | | /// sequence AND stamps the draw with the server clock in one atomic document update, so no |
| | | 659 | | /// delivery claim can interleave between the sequence draw and the clock read. |
| | | 660 | | /// <c>$ifNull</c> seeds a fresh counter document at 1, matching the delivery claim's |
| | | 661 | | /// <c>$inc</c> upsert. |
| | | 662 | | /// </summary> |
| | | 663 | | internal static UpdateDefinition<BsonDocument> BuildSubscriptionStartPipeline() |
| | 395 | 664 | | => Builders<BsonDocument>.Update.Pipeline(new[] |
| | 395 | 665 | | { |
| | 395 | 666 | | new BsonDocument("$set", new BsonDocument |
| | 395 | 667 | | { |
| | 395 | 668 | | ["seq"] = new BsonDocument("$add", new BsonArray { new BsonDocument("$ifNull", new BsonArray { "$seq", 0 |
| | 395 | 669 | | ["drawn_at"] = "$$NOW" |
| | 395 | 670 | | }) |
| | 395 | 671 | | }); |
| | | 672 | | |
| | | 673 | | public async Task<bool> IsMessageAcknowledgedAsync(Guid messageId, CancellationToken cancellationToken) |
| | | 674 | | { |
| | 118 | 675 | | await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); |
| | 118 | 676 | | var filter = Builders<MongoChannelMessageDocument>.Filter.Eq(item => item.Id, messageId) |
| | 118 | 677 | | & NotExpiredOnServerClock<MongoChannelMessageDocument>(); |
| | | 678 | | // Direct member projection, not an anonymous type: anonymous projections lower to the |
| | | 679 | | // RequiresUnreferencedCode Expression.New(ctor, args, members) overload, which the ILC |
| | | 680 | | // trim analysis rejects (Roslyn's analyzer skips compiler-lowered expression trees, so |
| | | 681 | | // only Native AOT publishes catch it). Missing document and unacked document both come |
| | | 682 | | // back as null, which is exactly the contract here. |
| | 118 | 683 | | var ackedAtUtc = await _messages.Find(filter) |
| | 118 | 684 | | .Project(item => item.AckedAtUtc) |
| | 118 | 685 | | .FirstOrDefaultAsync(cancellationToken).ConfigureAwait(false); |
| | 118 | 686 | | return ackedAtUtc is not null; |
| | 118 | 687 | | } |
| | | 688 | | |
| | | 689 | | public async Task UpsertSubscriberAsync(string correlationId, Guid registrationId, string instanceId, TimeSpan ttl, |
| | | 690 | | { |
| | 396 | 691 | | await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); |
| | 396 | 692 | | await _subscribers.UpdateOneAsync( |
| | 396 | 693 | | Builders<MongoChannelSubscriberDocument>.Filter.Eq(item => item.Id, RegistrationKey(correlationId, registrat |
| | 396 | 694 | | BuildSubscriberUpsertPipeline(correlationId, registrationId, instanceId, ttl), |
| | 396 | 695 | | new UpdateOptions { IsUpsert = true }, |
| | 396 | 696 | | cancellationToken).ConfigureAwait(false); |
| | 396 | 697 | | } |
| | | 698 | | |
| | | 699 | | /// <summary> |
| | | 700 | | /// Upsert pipeline for a subscriber liveness document, with <c>expires_at</c> computed on the |
| | | 701 | | /// server clock ($$NOW) — app-clock liveness math would let a skewed client look dead (or |
| | | 702 | | /// immortal) to publishers comparing against server-side expiry. |
| | | 703 | | /// </summary> |
| | | 704 | | internal static UpdateDefinition<MongoChannelSubscriberDocument> BuildSubscriberUpsertPipeline( |
| | | 705 | | string correlationId, |
| | | 706 | | Guid registrationId, |
| | | 707 | | string instanceId, |
| | | 708 | | TimeSpan ttl) |
| | 752 | 709 | | => Builders<MongoChannelSubscriberDocument>.Update.Pipeline(new[] |
| | 752 | 710 | | { |
| | 752 | 711 | | new BsonDocument("$set", new BsonDocument |
| | 752 | 712 | | { |
| | 752 | 713 | | ["correlation_id"] = correlationId, |
| | 752 | 714 | | ["registration_id"] = new BsonBinaryData(registrationId, GuidRepresentation.Standard), |
| | 752 | 715 | | ["instance_id"] = instanceId, |
| | 752 | 716 | | ["expires_at"] = new BsonDocument("$add", new BsonArray { "$$NOW", ttl.TotalMilliseconds }) |
| | 752 | 717 | | }) |
| | 752 | 718 | | }); |
| | | 719 | | |
| | | 720 | | public async Task HeartbeatSubscribersAsync( |
| | | 721 | | string instanceId, |
| | | 722 | | IReadOnlyCollection<(string CorrelationId, Guid RegistrationId)> registrations, |
| | | 723 | | TimeSpan ttl, |
| | | 724 | | CancellationToken cancellationToken) |
| | | 725 | | { |
| | 355 | 726 | | if (registrations.Count == 0) |
| | 2 | 727 | | return; |
| | | 728 | | |
| | 353 | 729 | | await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); |
| | | 730 | | |
| | | 731 | | // Per-registration upserts rather than one bare UpdateMany: the caller only heartbeats |
| | | 732 | | // registrations that are live in this process, so a missing document means the TTL reaper |
| | | 733 | | // deleted it (e.g. after a >timeout stall) — re-creating it here is what brings the waiter |
| | | 734 | | // back from "permanently invisible". Same document shape as UpsertSubscriberAsync. |
| | 353 | 735 | | var writes = new List<WriteModel<MongoChannelSubscriberDocument>>(registrations.Count); |
| | 1418 | 736 | | foreach (var (correlationId, registrationId) in registrations) |
| | | 737 | | { |
| | 356 | 738 | | writes.Add(new UpdateOneModel<MongoChannelSubscriberDocument>( |
| | 356 | 739 | | Builders<MongoChannelSubscriberDocument>.Filter.Eq(item => item.Id, RegistrationKey(correlationId, regis |
| | 356 | 740 | | BuildSubscriberUpsertPipeline(correlationId, registrationId, instanceId, ttl)) |
| | 356 | 741 | | { |
| | 356 | 742 | | IsUpsert = true |
| | 356 | 743 | | }); |
| | | 744 | | } |
| | | 745 | | |
| | 353 | 746 | | await _subscribers.BulkWriteAsync( |
| | 353 | 747 | | writes, |
| | 353 | 748 | | new BulkWriteOptions { IsOrdered = false }, |
| | 353 | 749 | | cancellationToken).ConfigureAwait(false); |
| | 343 | 750 | | } |
| | | 751 | | |
| | | 752 | | public async Task DeleteSubscriberAsync(string correlationId, Guid registrationId, CancellationToken cancellationTok |
| | | 753 | | { |
| | 442 | 754 | | await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); |
| | 442 | 755 | | await _subscribers.DeleteOneAsync( |
| | 442 | 756 | | Builders<MongoChannelSubscriberDocument>.Filter.Eq(item => item.Id, RegistrationKey(correlationId, registrat |
| | 442 | 757 | | cancellationToken).ConfigureAwait(false); |
| | 428 | 758 | | } |
| | | 759 | | |
| | | 760 | | public async Task<long> CountActiveSubscribersAsync(string correlationId, CancellationToken cancellationToken) |
| | | 761 | | { |
| | 565 | 762 | | await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); |
| | 565 | 763 | | var filter = Builders<MongoChannelSubscriberDocument>.Filter.Eq(item => item.CorrelationId, correlationId) |
| | 565 | 764 | | & NotExpiredOnServerClock<MongoChannelSubscriberDocument>(); |
| | 565 | 765 | | return await _subscribers.CountDocumentsAsync(filter, cancellationToken: cancellationToken).ConfigureAwait(false |
| | 557 | 766 | | } |
| | | 767 | | |
| | | 768 | | /// <summary> |
| | | 769 | | /// Server-clock expiry filter (<c>$expr: expires_at > $$NOW</c>): message, liveness, and |
| | | 770 | | /// recovery expiry are all stamped with $$NOW, so comparing them against the app clock would |
| | | 771 | | /// reintroduce the clock-skew hole the server-side stamps exist to close. |
| | | 772 | | /// </summary> |
| | | 773 | | internal static FilterDefinition<TDocument> NotExpiredOnServerClock<TDocument>() |
| | 7473 | 774 | | => new BsonDocument("$expr", new BsonDocument("$gt", new BsonArray { "$expires_at", "$$NOW" })); |
| | | 775 | | |
| | | 776 | | /// <summary> |
| | | 777 | | /// Watches the message collection with a change stream and invokes |
| | | 778 | | /// <paramref name="onNotification"/> with the correlation id of every inserted response. One |
| | | 779 | | /// stream serves every local waiter — the caller routes the id to the right subscription — so |
| | | 780 | | /// waiter count never multiplies server-side cursors. Runs until cancellation or a stream error. |
| | | 781 | | /// </summary> |
| | | 782 | | public async Task WatchMessagesAsync(Func<string?, Task> onNotification, CancellationToken cancellationToken) |
| | | 783 | | { |
| | 359 | 784 | | await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); |
| | 359 | 785 | | using var cursor = await _messages.WatchAsync( |
| | 359 | 786 | | BuildMessageWatchPipeline(), |
| | 359 | 787 | | new ChangeStreamOptions { FullDocument = ChangeStreamFullDocumentOption.UpdateLookup }, |
| | 359 | 788 | | cancellationToken).ConfigureAwait(false); |
| | 1231 | 789 | | while (await cursor.MoveNextAsync(cancellationToken).ConfigureAwait(false)) |
| | | 790 | | { |
| | 2692 | 791 | | foreach (var change in cursor.Current) |
| | 474 | 792 | | await onNotification(change.FullDocument?.CorrelationId).ConfigureAwait(false); |
| | | 793 | | } |
| | 2 | 794 | | } |
| | | 795 | | |
| | | 796 | | /// <summary> |
| | | 797 | | /// Change-stream pipeline for response wakes: a <c>$match</c> on insert events. The correlation |
| | | 798 | | /// id travels in the event's full document, letting the dispatcher scan only the signaled |
| | | 799 | | /// correlation id — the targeted-wake contract. |
| | | 800 | | /// </summary> |
| | | 801 | | internal static PipelineDefinition<ChangeStreamDocument<MongoChannelMessageDocument>, ChangeStreamDocument<MongoChan |
| | 361 | 802 | | => new EmptyPipelineDefinition<ChangeStreamDocument<MongoChannelMessageDocument>>() |
| | 361 | 803 | | .Match(change => change.OperationType == ChangeStreamOperationType.Insert); |
| | | 804 | | |
| | | 805 | | /// <summary>Returns <c>true</c> when the server rejected the change stream itself (not a transient cursor error).</ |
| | | 806 | | internal static bool IsChangeStreamUnsupported(Exception exception) |
| | 6 | 807 | | => exception is MongoCommandException commandException |
| | 6 | 808 | | && (commandException.Code == 40573 |
| | 6 | 809 | | || commandException.Message.Contains("only supported on replica sets", StringComparison.OrdinalIgnoreCase |
| | | 810 | | |
| | | 811 | | internal static string RegistrationKey(string correlationId, Guid registrationId) |
| | 1971 | 812 | | => $"{correlationId}:{registrationId:N}"; |
| | | 813 | | |
| | | 814 | | /// <summary> |
| | | 815 | | /// Name of the derived ack-counter collection. Part of the effective collection-name plan: |
| | | 816 | | /// options validation must keep the configured collections distinct from this derived name, |
| | | 817 | | /// or counter documents land in (for example) the TTL-indexed recovery collection, where the |
| | | 818 | | /// reaper would silently delete the ack sequence and reset the same-tick tie-breaker. |
| | | 819 | | /// </summary> |
| | 2975 | 820 | | internal static string CountersCollectionName(string messageCollection) => $"{messageCollection}_counters"; |
| | | 821 | | |
| | | 822 | | /// <summary>Validates a MongoDB collection name coming from options.</summary> |
| | | 823 | | public static void ValidateCollectionName(string? value, string name) |
| | | 824 | | { |
| | 3279 | 825 | | if (string.IsNullOrWhiteSpace(value)) |
| | 2 | 826 | | throw new InvalidOperationException($"{nameof(MongoDbAsyncResponseChannelOptions)}.{name} must be configured |
| | 3277 | 827 | | if (value.Contains('$') || value.Contains('\0') |
| | 3277 | 828 | | || value.StartsWith("system.", StringComparison.Ordinal) || value.Contains(".system.", StringComparison.Ordi |
| | 6 | 829 | | throw new InvalidOperationException( |
| | 6 | 830 | | $"{nameof(MongoDbAsyncResponseChannelOptions)}.{name} '{value}' must be a valid MongoDB collection name |
| | 3271 | 831 | | if (string.Equals(value, MongoOwnershipLedger.CollectionName, StringComparison.Ordinal)) |
| | 0 | 832 | | throw new InvalidOperationException( |
| | 0 | 833 | | $"{nameof(MongoDbAsyncResponseChannelOptions)}.{name} '{value}' is reserved for the cross-component owne |
| | 3271 | 834 | | } |
| | | 835 | | |
| | | 836 | | /// <summary>Disposes the Mongo client when the store created (and therefore owns) it.</summary> |
| | | 837 | | public void Dispose() |
| | | 838 | | { |
| | 381 | 839 | | _ensureGate.Dispose(); |
| | 381 | 840 | | (_ownedClient as IDisposable)?.Dispose(); |
| | 0 | 841 | | } |
| | | 842 | | } |
| | | 843 | | |
| | | 844 | | internal sealed class MongoRecoveryStateDocument |
| | | 845 | | { |
| | | 846 | | [BsonId] |
| | | 847 | | [BsonElement("_id")] |
| | | 848 | | public string Id { get; set; } = ""; |
| | | 849 | | |
| | | 850 | | [BsonElement("correlation_id")] |
| | | 851 | | public string CorrelationId { get; set; } = ""; |
| | | 852 | | |
| | | 853 | | [BsonElement("registration_id")] |
| | | 854 | | [BsonGuidRepresentation(GuidRepresentation.Standard)] |
| | | 855 | | public Guid RegistrationId { get; set; } |
| | | 856 | | |
| | | 857 | | [BsonElement("state_json")] |
| | | 858 | | public string StateJson { get; set; } = ""; |
| | | 859 | | |
| | | 860 | | [BsonElement("expires_at")] |
| | | 861 | | public DateTime ExpiresAtUtc { get; set; } |
| | | 862 | | |
| | | 863 | | [BsonElement("registered_at")] |
| | | 864 | | public DateTime RegisteredAtUtc { get; set; } |
| | | 865 | | } |
| | | 866 | | |
| | | 867 | | internal sealed class MongoChannelMessageDocument |
| | | 868 | | { |
| | | 869 | | [BsonId] |
| | | 870 | | [BsonElement("_id")] |
| | | 871 | | [BsonGuidRepresentation(GuidRepresentation.Standard)] |
| | | 872 | | public Guid Id { get; set; } |
| | | 873 | | |
| | | 874 | | [BsonElement("correlation_id")] |
| | | 875 | | public string CorrelationId { get; set; } = ""; |
| | | 876 | | |
| | | 877 | | /// <summary>Null only on a sweep projection of an acknowledged document (<see cref="MongoDbChannelStore.SweepProjec |
| | | 878 | | [BsonElement("envelope_json")] |
| | | 879 | | public string? EnvelopeJson { get; set; } = ""; |
| | | 880 | | |
| | | 881 | | [BsonElement("created_at")] |
| | | 882 | | public DateTime CreatedAtUtc { get; set; } |
| | | 883 | | |
| | | 884 | | [BsonElement("expires_at")] |
| | | 885 | | public DateTime ExpiresAtUtc { get; set; } |
| | | 886 | | |
| | | 887 | | [BsonElement("acked_at")] |
| | | 888 | | public DateTime? AckedAtUtc { get; set; } |
| | | 889 | | |
| | | 890 | | [BsonElement("acked_seq")] |
| | | 891 | | public long? AckedSeq { get; set; } |
| | | 892 | | |
| | | 893 | | [BsonElement("recovery_claimed")] |
| | | 894 | | public bool RecoveryClaimed { get; set; } |
| | | 895 | | } |
| | | 896 | | |
| | | 897 | | internal sealed class MongoChannelSubscriberDocument |
| | | 898 | | { |
| | | 899 | | [BsonId] |
| | | 900 | | [BsonElement("_id")] |
| | | 901 | | public string Id { get; set; } = ""; |
| | | 902 | | |
| | | 903 | | [BsonElement("correlation_id")] |
| | | 904 | | public string CorrelationId { get; set; } = ""; |
| | | 905 | | |
| | | 906 | | [BsonElement("registration_id")] |
| | | 907 | | [BsonGuidRepresentation(GuidRepresentation.Standard)] |
| | | 908 | | public Guid RegistrationId { get; set; } |
| | | 909 | | |
| | | 910 | | [BsonElement("instance_id")] |
| | | 911 | | public string InstanceId { get; set; } = ""; |
| | | 912 | | |
| | | 913 | | [BsonElement("expires_at")] |
| | | 914 | | public DateTime ExpiresAtUtc { get; set; } |
| | | 915 | | } |